feat(notification): digest aggregates, supplement-impact batch, and free-collab hooks

Add notification digests with ON CONFLICT DO UPDATE, migrations 00068-00070,
supplementimpact module (analyze → confirm → claim-aware split), and wire
coordinator/mutate hooks plus proposal_required digests.
This commit is contained in:
weichen
2026-09-24 11:44:20 +00:00
parent ebb36fb85c
commit 4e0e6a1418
29 changed files with 1546 additions and 18 deletions
@@ -0,0 +1,22 @@
# 自由协作 + 认领提醒 — 任务记录
- **task_id**: free-collab-claim-notify-2026-09-24
- **status**: in_progress
- **owner**: Cursor
- **mutex**: notification+supplementimpact+events+docs
- **companion_task**: non-author-member-access-2026-09-24 (server+docgov+web)
- **base_dev_sha**: ebb36fb85c80b83e895b29df5d9056c68f5efe31
- **design**: [../system-design/2026-09-24-free-collab-claim-notify-design.md](../system-design/2026-09-24-free-collab-claim-notify-design.md)
- **write_scope**: internal/{notification,events,app,bizdev,coordinator,traceimpact,supplementimpact}, deploy/postgres, .doc
- **review_state**: pending
- **integration_state**: pending
- **deployment**: not_started
- **cleanup**: pending
- **remote_sync**: not_requested
## 批次
- A 自由协作通道(companion worktree)
- B 认领补齐 + 互斥加固(companion)
- C 提醒机制 + 收件箱
- D 补充影响批次 + 确认卡 + 分流执行
@@ -0,0 +1,58 @@
# 自由协作通道 + 认领态分流 + 文档变更提醒机制
**status**: in_progress
**task_id**: free-collab-claim-notify-2026-09-24
**companion**: non-author-member-access-2026-09-24(持有 server/docgov/web 写租约)
**base_dev_sha**: ebb36fb85c80b83e895b29df5d9056c68f5efe31
**授权**: 用户确认方案后实施(会话 2026-09-24)
## 1. 目标
1. 任意项目成员随时自由对话、上传、总结(自由协作通道)
2. 补充材料可按认领机制更新涉及文档(确认卡勾选)
3. 已认领须编制负责人确认并由负责人发起修改;未认领只发摘要提醒
4. 单文档同时只被一个任务修改;冲突提示且不执行
## 2. 认领语义(不变)
- 非模板(`is_template_only=false`)+ 编制负责人查看或编辑 → `author_claimed=true`
- 模板态查看不算认领
- 已认领 → `RequiresProposal`;未认领 → `AllowedDirectEdit`
- 补齐:正文读取成功后自动 `RecordDocumentViewed`(author);claim/proposal 路由补项目鉴权
## 3. 自由协作通道
- 准入 = 项目成员(`RequireProjectAccess`),不查阶段/duty/实例状态
- 适用:chat/help/query/impact.analyze/supplement/上传/总结
- 受控文档写动作仍走 duty+阶段+认领
- `allowActiveStageChatAccess` 仅约束读取他人阶段对话/受控正文,不约束发起自由对话
## 4. 补充 → 影响 → 确认卡 → 分流
- `project.fact.supplement` 成功后 → `traceimpact.Analyze` → `supplement_impact_batches/items`
- 卡片 `SUPPLEMENT_IMPACT_CONFIRM`;决策 API 用 nonce+digest+revision CAS(不走 intentflow plan tier)
- 未认领:直改 + info 提醒;已认领:提案 + action_required;busy:skipped_busy 不回滚
- 批准后由负责人身份 `ForceDirectEdit` + `MarkProposalApplied`
## 5. 提醒聚合
- `notification_digests`:`(user_id, digest_kind, doc_id, window_date)` UNIQUE,ON CONFLICT DO UPDATE
- `notifications.severity` ∈ {action_required, info}
- 新事件 `document.proposal_required`
- 文案:见实施计划 §4.5(单次/同日多次/多文档扇出)
## 6. 互斥加固
- `persistLeaseLocked` 条件更新;heartbeat/release holder 归属鉴权
- fencing token 接入 HTTP;`LeaseHolderDisplay` 可读文案
## 7. 迁移
- 00068 notifications digest_id + severity
- 00069 notification_digests
- 00070 supplement_impact_batches + items
## 8. 红线
- 不放宽 is_locked / duty 写鉴权;不升 MULTI_INTENT_PLAN_TIER;不物理删 record 表
- 索引缺失不得显示为无影响
+9
View File
@@ -7,6 +7,15 @@
## 📋 报告清单
### 0b. 自由协作与认领提醒(2026-09-24)
| 项 | 链接 |
|-----|------|
| 主设计 | [2026-09-24-free-collab-claim-notify-design.md](2026-09-24-free-collab-claim-notify-design.md) |
| 任务 | [../ops/short-term/2026-09-24-free-collab-claim-notify.md](../ops/short-term/2026-09-24-free-collab-claim-notify.md) |
| 一句话总结 | 自由协作通道 + 认领态分流更新 + 通知 digest 聚合 |
### 0. T20 加固设计(2026-09-22)
| 项 | 链接 |
+26
View File
@@ -39,6 +39,7 @@ imports:
- kb
- materialindex
- maturity
- notification
- projectcontext
- projectworkflow
- quote
@@ -114,6 +115,7 @@ imports:
- projectinstance
- projectworkflow
- storage
- supplementimpact
- sysadmin
- taskrun
- traceimpact
@@ -307,6 +309,10 @@ imports:
sysadmin:
- config
traceimpact: []
supplementimpact:
- docgov
- notification
- traceimpact
table_owners:
document_records:
@@ -375,6 +381,12 @@ table_owners:
- app/docproject
notifications:
- notification
notification_digests:
- notification
supplement_impact_batches:
- supplementimpact
supplement_impact_items:
- supplementimpact
stage_advance_requests:
- notification
final_artifacts:
@@ -466,6 +478,16 @@ event_producers:
intent.plan.completed:
- app/intentflow
- events
document.proposal_required:
- docgov
- bizdev
- app/supplementimpact
- events
document.unclaimed_edit_applied:
- docgov
- bizdev
- app/supplementimpact
- events
event_consumers:
outbox.notify:
@@ -499,6 +521,10 @@ event_consumers:
intent.plan.completed:
- projection
- notify
document.proposal_required:
- notify
document.unclaimed_edit_applied:
- notify
unwrap_owners:
- docgov
@@ -0,0 +1,16 @@
-- +goose Up
ALTER TABLE notifications
ADD COLUMN IF NOT EXISTS digest_id BIGINT,
ADD COLUMN IF NOT EXISTS severity TEXT NOT NULL DEFAULT 'info';
DO $$ BEGIN
ALTER TABLE notifications
ADD CONSTRAINT chk_notification_severity CHECK (severity IN ('action_required', 'info'));
EXCEPTION WHEN duplicate_object THEN NULL;
END $$;
CREATE INDEX IF NOT EXISTS idx_notifications_digest_id ON notifications (digest_id) WHERE digest_id IS NOT NULL;
-- +goose Down
-- record retention: do not drop columns
SELECT 1;
@@ -0,0 +1,25 @@
-- +goose Up
CREATE TABLE IF NOT EXISTS notification_digests (
id BIGSERIAL PRIMARY KEY,
user_id BIGINT NOT NULL REFERENCES users(id),
digest_kind TEXT NOT NULL,
doc_id TEXT NOT NULL DEFAULT '',
window_date DATE NOT NULL,
occurrence_count INT NOT NULL DEFAULT 1,
last_actor_id BIGINT,
last_actor_name TEXT NOT NULL DEFAULT '',
last_at TIMESTAMPTZ NOT NULL DEFAULT now(),
item_summaries JSONB NOT NULL DEFAULT '[]'::jsonb,
truncated_count INT NOT NULL DEFAULT 0,
notification_id BIGINT REFERENCES notifications(id) ON DELETE SET NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
CONSTRAINT uq_notification_digest_window UNIQUE (user_id, digest_kind, doc_id, window_date)
);
CREATE INDEX IF NOT EXISTS idx_notification_digests_user_kind
ON notification_digests (user_id, digest_kind, window_date DESC);
-- +goose Down
-- record retention: do not drop table
SELECT 1;
@@ -0,0 +1,49 @@
-- +goose Up
CREATE TABLE IF NOT EXISTS supplement_impact_batches (
batch_id TEXT PRIMARY KEY,
project_id BIGINT NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
session_id TEXT NOT NULL DEFAULT '',
initiator_user_id BIGINT NOT NULL REFERENCES users(id),
source_kind TEXT NOT NULL DEFAULT 'fact_supplement',
source_summary TEXT NOT NULL DEFAULT '',
coverage_status TEXT NOT NULL DEFAULT 'index_pending',
status TEXT NOT NULL DEFAULT 'pending',
decision_nonce TEXT NOT NULL DEFAULT '',
decision_digest TEXT NOT NULL DEFAULT '',
revision BIGINT NOT NULL DEFAULT 1,
expires_at TIMESTAMPTZ NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
CONSTRAINT chk_sib_status CHECK (status IN ('pending','decided','running','done','partially_done','expired','cancelled'))
);
CREATE TABLE IF NOT EXISTS supplement_impact_items (
id BIGSERIAL PRIMARY KEY,
batch_id TEXT NOT NULL REFERENCES supplement_impact_batches(batch_id) ON DELETE CASCADE,
doc_id TEXT NOT NULL,
title TEXT NOT NULL DEFAULT '',
item_key TEXT NOT NULL DEFAULT '',
impact_type TEXT NOT NULL DEFAULT '',
depth INT NOT NULL DEFAULT 0,
reason TEXT NOT NULL DEFAULT '',
author_claimed_snapshot BOOLEAN NOT NULL DEFAULT FALSE,
author_id_snapshot BIGINT NOT NULL DEFAULT 0,
busy_snapshot BOOLEAN NOT NULL DEFAULT FALSE,
change_summary TEXT NOT NULL DEFAULT '',
selected BOOLEAN NOT NULL DEFAULT FALSE,
outcome TEXT NOT NULL DEFAULT '',
proposal_id TEXT NOT NULL DEFAULT '',
run_id TEXT NOT NULL DEFAULT '',
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
CONSTRAINT chk_sii_outcome CHECK (
outcome = '' OR outcome IN ('applied','proposal_pending','proposal_rejected','skipped_busy','skipped_locked','failed')
)
);
CREATE INDEX IF NOT EXISTS idx_sii_batch ON supplement_impact_items (batch_id);
CREATE INDEX IF NOT EXISTS idx_sib_project ON supplement_impact_batches (project_id, created_at DESC);
-- +goose Down
-- record retention: do not drop tables
SELECT 1;
+6 -1
View File
@@ -31,6 +31,7 @@ import (
"chat-agent-system/internal/jevdecide"
"chat-agent-system/internal/kb"
"chat-agent-system/internal/notification"
"chat-agent-system/internal/supplementimpact"
"chat-agent-system/internal/observability"
"chat-agent-system/internal/projectinstance"
"chat-agent-system/internal/projectworkflow"
@@ -75,6 +76,8 @@ type Runtime struct {
Gov *docgov.GovernanceEngine
Rev *docrevision.Service
Trace *traceimpact.Service
SupplementImpact *supplementimpact.Service
mutateOwner *MutateOwner
Ctx *ctxmgr.ContextManager
CAS *storage.CAS
FreeDoc *freedoc.Service
@@ -312,7 +315,8 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) {
pflags := intentflow.CurrentFeatureFlags()
liveOwners := intentflow.LiveOwners{}
if pflags.U10Mutate {
liveOwners.Mutate = NewMutateOwner(projectAccess(rt), projectBlackboardPort(pool, parts.Ctx))
rt.mutateOwner = NewMutateOwner(projectAccess(rt), projectBlackboardPort(pool, parts.Ctx))
liveOwners.Mutate = rt.mutateOwner
}
if pflags.U11DocGov {
liveOwners.DocGov = NewDocGovOwner(parts.Gov, projectAccess(rt), nil)
@@ -335,6 +339,7 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) {
svc.SetAgentMaterializer(rt.Docflow)
rt.bindDocumentNotify()
rt.bindWorkflowApprovalHook()
rt.wireSupplementImpact()
return rt, nil
}
+4
View File
@@ -85,6 +85,10 @@ func notifyCopy(eventName string) (kind, title, body string) {
return notification.TypeDocChangeReportDegraded, "变更报告未完整评估", "变更报告降级,受控文档暂不可送审。"
case events.EventApproved:
return notification.TypeDocApproved, "文档已批准", "文档已完成终审批准。"
case events.EventDocProposalRequired:
return notification.TypeDocProposalRequired, "文档变更待确认", "已认领文档有变更提案,请确认后由您发起修改。"
case events.EventDocUnclaimedEdit:
return notification.TypeDocUnclaimedEditApplied, "文档已更新", "未认领文档已被直接更新(无需您确认)。"
default:
return "", "", ""
}
+17
View File
@@ -23,6 +23,8 @@ type MutateOwner struct {
Access ProjectAccess
Blackboard ProjectBlackboardPort
Denied bool
// OnSupplement is optional; composition typically wires supplementimpact.TriggerAsync.
OnSupplement func(ctx context.Context, projectID, userID int64, sessionID, summary, markdown string)
mu sync.Mutex
receipts map[string]intentflow.Receipt
@@ -91,6 +93,21 @@ func (m *MutateOwner) Execute(ctx context.Context, exec intentflow.StepExec) (in
return intentflow.Receipt{}, err
}
newVer = got
if m.OnSupplement != nil {
summary := strings.TrimSpace(exec.SafeSummary)
if summary == "" {
summary = "project facts supplemented"
}
markdown := summary
if len(facts) > 0 {
parts := make([]string, 0, len(facts))
for k, v := range facts {
parts = append(parts, k+": "+v)
}
markdown = strings.Join(parts, "\n")
}
m.OnSupplement(ctx, exec.Principal.ProjectID, exec.Principal.UserID, exec.Principal.SessionID, summary, markdown)
}
}
rec := intentflow.Receipt{
+72
View File
@@ -0,0 +1,72 @@
package app
import (
"context"
"fmt"
"log/slog"
"time"
"chat-agent-system/internal/intent"
"chat-agent-system/internal/supplementimpact"
)
func (r *Runtime) wireSupplementImpact() {
if r == nil || r.Pool == nil || r.Gov == nil || r.Trace == nil {
return
}
if r.SupplementImpact != nil {
return
}
svc := &supplementimpact.Service{
Store: supplementimpact.NewPGStore(r.Pool),
Docs: r.Gov,
Analyze: r.Trace,
Notify: r.Notify,
Propose: r.Gov,
}
r.SupplementImpact = svc
if r.Biz != nil && r.Notify != nil {
r.Biz.SetDigestNotifier(r.Notify)
}
if r.Engine != nil {
r.Engine.SetOnSupplement(func(ctx context.Context, projectID, userID int64, sessionID, summary, markdown string) {
b, err := svc.CreateFromSupplement(ctx, projectID, userID, sessionID, summary, markdown)
if err != nil {
slog.Warn("supplementimpact sync create failed", "error", err, "project_id", projectID)
return
}
r.Engine.SetPendingSupplementCard(batchToCard(b))
})
}
if r.mutateOwner != nil {
r.mutateOwner.OnSupplement = func(ctx context.Context, projectID, userID int64, sessionID, summary, markdown string) {
supplementimpact.TriggerAsync(svc, projectID, userID, sessionID, summary, markdown)
}
}
}
func batchToCard(b *supplementimpact.Batch) *intent.InteractiveCardPayload {
if b == nil {
return nil
}
title, summary := supplementimpact.BuildConfirmCardMeta(b)
actions := []intent.CardAction{
{ActionID: "confirm_supplement_impact", Label: "确认更新所选文档", Style: "primary", IsAffirmativeDefault: true},
{ActionID: "cancel_supplement_impact", Label: "取消", Style: "subtle"},
}
body := summary + "\n" + fmt.Sprintf("batch_id=%s\nnonce=%s\nrevision=%d\ncoverage=%s", b.BatchID, b.DecisionNonce, b.Revision, b.CoverageStatus)
for _, it := range b.Items {
body += fmt.Sprintf("\n- %s | claimed=%v busy=%v | %s", it.DocID, it.AuthorClaimedSnapshot, it.BusySnapshot, it.Title)
}
return &intent.InteractiveCardPayload{
CardID: b.BatchID,
CardType: "SUPPLEMENT_IMPACT_CONFIRM",
Title: title,
ProposedContent: body,
ImpactRisk: b.CoverageStatus,
CurrentStatus: "active",
TTLTurns: 8,
CreatedAt: time.Now().UTC(),
Actions: actions,
}
}
+50
View File
@@ -0,0 +1,50 @@
package bizdev
import (
"context"
"fmt"
"chat-agent-system/internal/docgov"
)
// TaskExecutor is the ForceDirectEdit execution surface used by AfterProposalApproved.
type TaskExecutor interface {
ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRunResult, error)
}
// AfterProposalApproved starts a ForceDirectEdit task run as the author after a
// proposal reaches approved_by_author. Finalize already calls MarkProposalApplied
// when FindApprovedProposal succeeds under ForceDirectEdit.
//
// Server composition (non-author worktree) should call this after
// DecideModificationProposal(approve=true) succeeds; pass a base TaskRunRequest
// with TaskKey/DocID/project fields filled from the document context.
func AfterProposalApproved(ctx context.Context, exec TaskExecutor, prop *docgov.ChangeProposal, authorID int64, base TaskRunRequest) (*TaskRunResult, error) {
if exec == nil {
return nil, fmt.Errorf("bizdev: AfterProposalApproved requires executor")
}
if prop == nil {
return nil, fmt.Errorf("bizdev: AfterProposalApproved requires proposal")
}
if authorID <= 0 {
return nil, fmt.Errorf("bizdev: AfterProposalApproved requires author id")
}
req := base
req.ForceDirectEdit = true
req.UserID = authorID
if req.AuthorID == 0 {
req.AuthorID = authorID
}
if req.DocID == "" {
req.DocID = prop.DocID
}
if req.ProjectID == 0 {
req.ProjectID = prop.ProjectID
}
return exec.ExecuteTask(ctx, req)
}
// AfterProposalApproved is the Service convenience wrapper.
func (s *Service) AfterProposalApproved(ctx context.Context, prop *docgov.ChangeProposal, authorID int64, base TaskRunRequest) (*TaskRunResult, error) {
return AfterProposalApproved(ctx, s, prop, authorID, base)
}
+63
View File
@@ -0,0 +1,63 @@
package bizdev_test
import (
"context"
"testing"
"chat-agent-system/internal/bizdev"
"chat-agent-system/internal/docgov"
)
type stubTaskExec struct {
last bizdev.TaskRunRequest
res *bizdev.TaskRunResult
err error
}
func (s *stubTaskExec) ExecuteTask(_ context.Context, req bizdev.TaskRunRequest) (*bizdev.TaskRunResult, error) {
s.last = req
return s.res, s.err
}
func TestAfterProposalApproved_ForceEditAsAuthor(t *testing.T) {
exec := &stubTaskExec{res: &bizdev.TaskRunResult{Success: true}}
prop := &docgov.ChangeProposal{
ProposalID: "prop_1",
DocID: "doc_a",
ProjectID: 42,
Status: docgov.ProposalApprovedByAuthor,
}
res, err := bizdev.AfterProposalApproved(context.Background(), exec, prop, 9, bizdev.TaskRunRequest{
TaskKey: "feasibility_reg",
ProjectName: "P",
Backend: "stub",
})
if err != nil {
t.Fatal(err)
}
if res == nil || !res.Success {
t.Fatalf("want success result, got %+v", res)
}
if !exec.last.ForceDirectEdit {
t.Fatal("expected ForceDirectEdit=true")
}
if exec.last.UserID != 9 || exec.last.AuthorID != 9 {
t.Fatalf("want author identity 9, got user=%d author=%d", exec.last.UserID, exec.last.AuthorID)
}
if exec.last.DocID != "doc_a" || exec.last.ProjectID != 42 {
t.Fatalf("want doc/project from proposal, got doc=%s project=%d", exec.last.DocID, exec.last.ProjectID)
}
}
func TestAfterProposalApproved_NilSafe(t *testing.T) {
if _, err := bizdev.AfterProposalApproved(context.Background(), nil, &docgov.ChangeProposal{}, 1, bizdev.TaskRunRequest{}); err == nil {
t.Fatal("nil executor must error")
}
exec := &stubTaskExec{}
if _, err := bizdev.AfterProposalApproved(context.Background(), exec, nil, 1, bizdev.TaskRunRequest{}); err == nil {
t.Fatal("nil proposal must error")
}
if _, err := bizdev.AfterProposalApproved(context.Background(), exec, &docgov.ChangeProposal{DocID: "d"}, 0, bizdev.TaskRunRequest{}); err == nil {
t.Fatal("author id 0 must error")
}
}
+16
View File
@@ -20,6 +20,7 @@ import (
"chat-agent-system/internal/freedoc"
"chat-agent-system/internal/kb"
"chat-agent-system/internal/materialindex"
"chat-agent-system/internal/notification"
"chat-agent-system/internal/projectworkflow"
"chat-agent-system/internal/squad"
"chat-agent-system/internal/storage"
@@ -53,6 +54,13 @@ type Service struct {
completedMainHook func(ctx context.Context, projectID int64) (bool, error)
// instanceDuty 可选:活跃工作流实例对 taskKey 的治理职责(由 app 装配,避免导入 projectinstance)。
instanceDuty InstanceDutySource
// notify 可选:提案拦截时写入 document digest(composition 注入)。
notify DigestNotifier
}
// DigestNotifier is the subset of notification.Service used on proposal_required.
type DigestNotifier interface {
UpsertDocumentDigest(ctx context.Context, p notification.UpsertDigestParams) error
}
// InstanceDutySource 返回项目当前活跃实例对 taskKey 的治理结论。
@@ -69,6 +77,14 @@ func (s *Service) SetInstanceDutySource(src InstanceDutySource) {
s.instanceDuty = src
}
// SetDigestNotifier wires optional claim-aware digest notifications.
func (s *Service) SetDigestNotifier(n DigestNotifier) {
if s == nil {
return
}
s.notify = n
}
// NewService 构造业务服务;生产/test 环境数据库不可用且未显式 ALLOW_IN_MEMORY 时返回 error。
func NewService(cfg *config.Config) (*Service, error) {
if cfg == nil {
+15
View File
@@ -13,6 +13,7 @@ import (
"chat-agent-system/internal/contracts/vocabulary"
"chat-agent-system/internal/docduty"
"chat-agent-system/internal/docgov"
"chat-agent-system/internal/notification"
"chat-agent-system/internal/docrevision"
"chat-agent-system/internal/kb"
"chat-agent-system/internal/projectcontext"
@@ -485,6 +486,20 @@ func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRun
prop, _ = s.docGov.ProposeModification(ctx, docID, "agent_squad", proposedChanges, impactAnalysis, riskAssessment)
}
if s.notify != nil && prop != nil && docRec.AuthorID > 0 {
_ = s.notify.UpsertDocumentDigest(ctx, notification.UpsertDigestParams{
UserID: docRec.AuthorID,
ProjectID: req.ProjectID,
DigestKind: notification.DigestKindProposalRequired,
DocID: docID,
DocTitle: artifactTitle,
Severity: notification.SeverityActionRequired,
ActorID: req.UserID,
ActorName: "agent_squad",
Summary: prop.ProposedChanges,
ProposalID: prop.ProposalID,
})
}
streamer := stream.FromContext(ctx)
if streamer != nil {
streamer.EmitGovernance(stream.GovernancePayload{
+27
View File
@@ -76,6 +76,10 @@ type CoordinatorEngine struct {
kbClient *kb.Client
impact ImpactActions
projectAccess ProjectAccess
// OnSupplement is optional; composition typically wires supplementimpact.TriggerAsync.
// Nil-safe: unset hooks are skipped.
OnSupplement func(ctx context.Context, projectID, userID int64, sessionID, summary, markdown string)
pendingSupplementCard *intent.InteractiveCardPayload
}
// StageTaskRunner is the application owner for ExecuteTask / ExecuteStage.
@@ -126,6 +130,29 @@ func (e *CoordinatorEngine) SetImpactActions(a ImpactActions) {
}
}
// SetOnSupplement wires the optional post-blackboard supplement impact hook.
func (e *CoordinatorEngine) SetOnSupplement(fn func(ctx context.Context, projectID, userID int64, sessionID, summary, markdown string)) {
if e != nil {
e.OnSupplement = fn
}
}
func (e *CoordinatorEngine) SetPendingSupplementCard(card *intent.InteractiveCardPayload) {
if e == nil {
return
}
e.pendingSupplementCard = card
}
func (e *CoordinatorEngine) TakePendingSupplementCard() *intent.InteractiveCardPayload {
if e == nil {
return nil
}
c := e.pendingSupplementCard
e.pendingSupplementCard = nil
return c
}
func NewCoordinatorEngine(runner StageTaskRunner) *CoordinatorEngine {
return &CoordinatorEngine{
stageTasks: runner,
+15
View File
@@ -85,6 +85,16 @@ func (e *CoordinatorEngine) handleSupplementInfo(cand *intent.CandidateIntent, u
}
}
if e != nil && e.OnSupplement != nil {
projectID := parseDecimalID(userCtx.Project.ID)
userID := parseDecimalID(userCtx.User.ID)
markdown := strings.TrimSpace(rawDoc)
if markdown == "" && factKey != "" && factVal != "" {
markdown = factKey + ": " + factVal
}
e.OnSupplement(context.Background(), projectID, userID, "", summaryKey, markdown)
}
nextExp := BuildNextExpectation(userCtx.Project.CurrentStage, userCtx.Project.Name)
userCtx.applyPending(nextExp)
@@ -105,6 +115,10 @@ func (e *CoordinatorEngine) handleSupplementInfo(cand *intent.CandidateIntent, u
sb.WriteString("已收录补充材料,并关联到项目事实黑板。\n后续文档修订与阶段推进时,我会自动引用这部分事实。")
}
if card := e.TakePendingSupplementCard(); card != nil {
userCtx.applyCard(card)
sb.WriteString("\n\n已分析补充材料对文档的潜在影响,请在确认卡中勾选要更新的文档。")
}
return &CoordinationResult{
Status: StatusAccepted,
Intent: cand.Intent,
@@ -113,6 +127,7 @@ func (e *CoordinatorEngine) handleSupplementInfo(cand *intent.CandidateIntent, u
AdvisorBrief: userCtx.LatestAdvisorBrief,
UpdatedContext: userCtx,
NextExpectation: nextExp,
InteractiveCard: userCtx.InteractiveCard,
}, nil
}
+4
View File
@@ -23,6 +23,8 @@ const (
EventIntentPlanAwaiting = "intent.plan.awaiting_input"
EventIntentPlanStepChanged = "intent.plan.step_changed"
EventIntentPlanCompleted = "intent.plan.completed"
EventDocProposalRequired = "document.proposal_required"
EventDocUnclaimedEdit = "document.unclaimed_edit_applied"
)
// Contract is the static allowlist row for one event name.
@@ -50,6 +52,8 @@ func StaticContracts() []Contract {
{EventName: EventIntentPlanAwaiting, SchemaVersion: 1, Enabled: true, Producers: []string{"app/intentflow", "events"}, Consumers: []string{ConsumerProjection, ConsumerNotify}, PayloadFields: []string{"plan_id", "revision", "status", "environment", "project_id", "user_id", "session_id", "summary_ref"}},
{EventName: EventIntentPlanStepChanged, SchemaVersion: 1, Enabled: true, Producers: []string{"app/intentflow", "events"}, Consumers: []string{ConsumerProjection, ConsumerNotify}, PayloadFields: []string{"plan_id", "revision", "step_id", "status", "environment", "project_id", "user_id", "session_id", "summary_ref"}},
{EventName: EventIntentPlanCompleted, SchemaVersion: 1, Enabled: true, Producers: []string{"app/intentflow", "events"}, Consumers: []string{ConsumerProjection, ConsumerNotify}, PayloadFields: []string{"plan_id", "revision", "status", "environment", "project_id", "user_id", "session_id", "summary_ref"}},
{EventName: EventDocProposalRequired, SchemaVersion: 1, Enabled: true, Producers: []string{"docgov", "bizdev", "app/supplementimpact", "events"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"doc_id", "proposal_id", "project_id"}},
{EventName: EventDocUnclaimedEdit, SchemaVersion: 1, Enabled: true, Producers: []string{"docgov", "bizdev", "app/supplementimpact", "events"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"doc_id", "project_id"}},
}
}
+186
View File
@@ -0,0 +1,186 @@
package notification
import (
"context"
"encoding/json"
"fmt"
"strings"
"time"
)
const (
DigestKindProposalRequired = "proposal_required"
DigestKindUnclaimedEdit = "unclaimed_edit"
SeverityActionRequired = "action_required"
SeverityInfo = "info"
maxDigestItems = 20
)
// DigestItem is one occurrence folded into a daily digest.
type DigestItem struct {
At time.Time `json:"at"`
ActorID int64 `json:"actor_id"`
ActorName string `json:"actor_name"`
Summary string `json:"summary"`
ProposalID string `json:"proposal_id,omitempty"`
ChangeCount int `json:"change_count,omitempty"`
}
// UpsertDigestParams drives ON CONFLICT DO UPDATE aggregation.
type UpsertDigestParams struct {
UserID int64
ProjectID int64
DigestKind string
DocID string
DocTitle string
Severity string
ActorID int64
ActorName string
Summary string
ProposalID string
ChangeCount int
WindowDate time.Time // truncated to date; zero → today UTC
}
// UpsertDocumentDigest aggregates same-day notifications per (user, kind, doc).
func (s *Service) UpsertDocumentDigest(ctx context.Context, p UpsertDigestParams) error {
if !s.Available() {
return ErrNotAvailable
}
if p.UserID <= 0 || p.DocID == "" || p.DigestKind == "" {
return fmt.Errorf("notification: digest requires user, doc, kind")
}
if p.Severity == "" {
p.Severity = SeverityInfo
}
window := p.WindowDate
if window.IsZero() {
now := time.Now().UTC()
window = time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC)
}
item := DigestItem{
At: time.Now().UTC(), ActorID: p.ActorID, ActorName: p.ActorName,
Summary: truncateRunes(p.Summary, 60), ProposalID: p.ProposalID, ChangeCount: p.ChangeCount,
}
itemJSON, _ := json.Marshal(item)
tx, err := s.pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
var digestID, notifID int64
var count int
err = tx.QueryRow(ctx, `
INSERT INTO notification_digests (
user_id, digest_kind, doc_id, window_date, occurrence_count,
last_actor_id, last_actor_name, last_at, item_summaries, truncated_count
) VALUES ($1,$2,$3,$4,1,$5,$6,now(), jsonb_build_array($7::jsonb), 0)
ON CONFLICT (user_id, digest_kind, doc_id, window_date) DO UPDATE SET
occurrence_count = notification_digests.occurrence_count + 1,
last_actor_id = EXCLUDED.last_actor_id,
last_actor_name = EXCLUDED.last_actor_name,
last_at = now(),
item_summaries = CASE
WHEN jsonb_array_length(notification_digests.item_summaries) >= $8 THEN notification_digests.item_summaries
ELSE notification_digests.item_summaries || EXCLUDED.item_summaries
END,
truncated_count = CASE
WHEN jsonb_array_length(notification_digests.item_summaries) >= $8
THEN notification_digests.truncated_count + 1
ELSE notification_digests.truncated_count
END,
updated_at = now()
RETURNING id, occurrence_count, COALESCE(notification_id, 0)
`, p.UserID, p.DigestKind, p.DocID, window, p.ActorID, p.ActorName, string(itemJSON), maxDigestItems).
Scan(&digestID, &count, &notifID)
if err != nil {
return err
}
title, body := digestCopy(p, count)
dedupeKey := fmt.Sprintf("digest:%s:%s:%s", p.DigestKind, p.DocID, window.Format("2006-01-02"))
notifType := typeForDigestKind(p.DigestKind)
if notifID == 0 {
err = tx.QueryRow(ctx, `
INSERT INTO notifications (user_id, project_id, type, title, body, dedupe_key, status, digest_id, severity)
VALUES ($1,$2,$3,$4,$5,$6,'unread',$7,$8)
ON CONFLICT (user_id, dedupe_key) DO UPDATE SET
title = EXCLUDED.title,
body = EXCLUDED.body,
status = 'unread',
severity = EXCLUDED.severity,
digest_id = EXCLUDED.digest_id
RETURNING id
`, p.UserID, p.ProjectID, notifType, title, body, dedupeKey, digestID, p.Severity).Scan(&notifID)
if err != nil {
return err
}
_, err = tx.Exec(ctx, `UPDATE notification_digests SET notification_id=$1 WHERE id=$2`, notifID, digestID)
if err != nil {
return err
}
} else {
_, err = tx.Exec(ctx, `
UPDATE notifications SET title=$1, body=$2, status='unread', severity=$3
WHERE id=$4 AND user_id=$5
`, title, body, p.Severity, notifID, p.UserID)
if err != nil {
return err
}
}
return tx.Commit(ctx)
}
func typeForDigestKind(kind string) string {
switch kind {
case DigestKindProposalRequired:
return TypeDocProposalRequired
case DigestKindUnclaimedEdit:
return TypeDocUnclaimedEditApplied
default:
return kind
}
}
func digestCopy(p UpsertDigestParams, count int) (title, body string) {
titleDoc := strings.TrimSpace(p.DocTitle)
if titleDoc == "" {
titleDoc = p.DocID
}
actor := strings.TrimSpace(p.ActorName)
if actor == "" {
actor = "成员"
}
summary := truncateRunes(p.Summary, 60)
switch p.DigestKind {
case DigestKindProposalRequired:
if count <= 1 {
return fmt.Sprintf("《%s》有变更待您确认", titleDoc),
fmt.Sprintf("%s 补充了「%s」,拟修改本文档。文档已被您认领,需您确认后由您发起修改。", actor, summary)
}
return fmt.Sprintf("《%s》累计 %d 次变更待您确认", titleDoc, count),
fmt.Sprintf("最近一次:%s。%d 次变更均来自他人补充材料,请逐份拍板。", actor, count)
case DigestKindUnclaimedEdit:
if count <= 1 {
return fmt.Sprintf("《%s》已被更新(无需您确认)", titleDoc),
fmt.Sprintf("%s 通过助手更新了本文档,来源:%s。本文档尚未被您认领,系统已直接应用。", actor, summary)
}
return fmt.Sprintf("《%s》今日已有 %d 次更新", titleDoc, count),
fmt.Sprintf("最近一次:%s;累计 %d 次更新。本文档尚未被您认领。", actor, count)
default:
return titleDoc, summary
}
}
func truncateRunes(s string, n int) string {
r := []rune(strings.TrimSpace(s))
if len(r) <= n {
return string(r)
}
return string(r[:n]) + "…"
}
// Ensure unused import guard for pgx when scanning with interfaces.
+4
View File
@@ -14,12 +14,16 @@ const (
EventDocHumanRevisionConflict = "document.human_revision_conflict"
EventDocDownstreamImpact = "document.downstream_impact_detected"
EventDocChangeReportDegraded = "document.change_report_degraded"
EventDocProposalRequired = "document.proposal_required"
EventDocUnclaimedEditApplied = "document.unclaimed_edit_applied"
TypeDocAgentCandidateReady = "document_agent_candidate_ready"
TypeDocAgentVersionCommitted = "document_agent_version_committed"
TypeDocDownstreamImpact = "document_downstream_impact"
TypeDocChangeReportDegraded = "document_change_report_degraded"
TypeDocApproved = "document_approved"
TypeDocProposalRequired = "document_proposal_required"
TypeDocUnclaimedEditApplied = "document_unclaimed_edit_applied"
)
// DocumentEventPayload is a minimal, permission-safe notification body (no markdown excerpts).
+44 -8
View File
@@ -173,6 +173,15 @@ func ignoreUnique(err error) error {
// ListProjectNotifications 列出项目下当前用户的通知(可选 since_id 增量)。
func (s *Service) ListProjectNotifications(ctx context.Context, userID, projectID int64, sinceID int64, limit int) ([]Notification, error) {
return s.listNotifications(ctx, userID, projectID, sinceID, limit, false)
}
// ListUserNotifications 列出当前用户跨项目通知;action_required 优先,再按 id 降序。
func (s *Service) ListUserNotifications(ctx context.Context, userID, sinceID int64, limit int) ([]Notification, error) {
return s.listNotifications(ctx, userID, 0, sinceID, limit, true)
}
func (s *Service) listNotifications(ctx context.Context, userID, projectID, sinceID int64, limit int, allProjects bool) ([]Notification, error) {
if !s.Available() {
return nil, ErrNotAvailable
}
@@ -181,18 +190,37 @@ func (s *Service) ListProjectNotifications(ctx context.Context, userID, projectI
}
query := `
SELECT n.id, n.user_id, n.project_id, COALESCE(n.event_id, 0), n.type, n.title, n.body, n.status, n.created_at,
COALESCE(e.payload, '{}'::jsonb)
SELECT n.id, n.user_id, n.project_id, COALESCE(n.event_id, 0), n.type, n.title, n.body, n.status,
COALESCE(n.severity, 'info'), COALESCE(n.digest_id, 0), n.created_at,
COALESCE(e.payload, '{}'::jsonb),
COALESCE(d.doc_id, ''),
COALESCE((
SELECT elem->>'proposal_id'
FROM jsonb_array_elements(COALESCE(d.item_summaries, '[]'::jsonb)) WITH ORDINALITY AS t(elem, ord)
WHERE COALESCE(elem->>'proposal_id', '') <> ''
ORDER BY ord DESC
LIMIT 1
), '')
FROM notifications n
LEFT JOIN outbox_events e ON e.id = n.event_id
WHERE n.user_id = $1 AND n.project_id = $2
LEFT JOIN notification_digests d ON d.id = n.digest_id
WHERE n.user_id = $1
`
args := []any{userID, projectID}
args := []any{userID}
if !allProjects {
query += ` AND n.project_id = $2`
args = append(args, projectID)
}
if sinceID > 0 {
query += ` AND n.id > $3`
query += ` AND n.id > $` + fmt.Sprint(len(args)+1)
args = append(args, sinceID)
}
query += ` ORDER BY n.id ASC LIMIT $` + fmt.Sprint(len(args)+1)
if allProjects {
query += ` ORDER BY CASE WHEN COALESCE(n.severity, 'info') = 'action_required' THEN 0 ELSE 1 END, n.id DESC`
} else {
query += ` ORDER BY n.id ASC`
}
query += ` LIMIT $` + fmt.Sprint(len(args)+1)
args = append(args, limit)
rows, err := s.pool.Query(ctx, query, args...)
@@ -205,15 +233,23 @@ func (s *Service) ListProjectNotifications(ctx context.Context, userID, projectI
for rows.Next() {
var n Notification
var payloadRaw []byte
var digestDocID, digestProposalID string
if err := rows.Scan(
&n.ID, &n.UserID, &n.ProjectID, &n.EventID, &n.Type, &n.Title, &n.Body, &n.Status, &n.CreatedAt,
&payloadRaw,
&n.ID, &n.UserID, &n.ProjectID, &n.EventID, &n.Type, &n.Title, &n.Body, &n.Status,
&n.Severity, &n.DigestID, &n.CreatedAt,
&payloadRaw, &digestDocID, &digestProposalID,
); err != nil {
return nil, err
}
if len(payloadRaw) > 0 {
_ = json.Unmarshal(payloadRaw, &n.Payload)
}
if n.Payload.DocID == "" && digestDocID != "" {
n.Payload.DocID = digestDocID
}
if n.Payload.ProposalID == "" && digestProposalID != "" {
n.Payload.ProposalID = digestProposalID
}
out = append(out, n)
}
return out, rows.Err()
+14 -9
View File
@@ -19,18 +19,23 @@ type StageReadyPayload struct {
ConfigRevision int `json:"config_revision"`
RequiredDocCount int `json:"required_doc_count"`
ApprovedDocCount int `json:"approved_doc_count"`
DocID string `json:"doc_id,omitempty"`
ProposalID string `json:"proposal_id,omitempty"`
BatchID string `json:"batch_id,omitempty"`
}
// Notification 用户站内通知。
type Notification struct {
ID int64 `json:"id"`
UserID int64 `json:"user_id"`
ProjectID int64 `json:"project_id"`
EventID int64 `json:"event_id,omitempty"`
Type string `json:"type"`
Title string `json:"title"`
Body string `json:"body"`
Status string `json:"status"`
CreatedAt time.Time `json:"created_at"`
ID int64 `json:"id"`
UserID int64 `json:"user_id"`
ProjectID int64 `json:"project_id"`
EventID int64 `json:"event_id,omitempty"`
Type string `json:"type"`
Title string `json:"title"`
Body string `json:"body"`
Status string `json:"status"`
Severity string `json:"severity,omitempty"`
DigestID int64 `json:"digest_id,omitempty"`
CreatedAt time.Time `json:"created_at"`
Payload StageReadyPayload `json:"payload,omitempty"`
}
+113
View File
@@ -0,0 +1,113 @@
package supplementimpact
import (
"context"
"fmt"
)
// HTTPBatchView is the JSON shape expected by server supplement-impact handlers.
type HTTPBatchView struct {
BatchID string `json:"batch_id"`
ProjectID int64 `json:"project_id"`
SessionID string `json:"session_id,omitempty"`
InitiatorUserID int64 `json:"initiator_user_id"`
SourceKind string `json:"source_kind,omitempty"`
SourceSummary string `json:"source_summary,omitempty"`
CoverageStatus string `json:"coverage_status,omitempty"`
Status string `json:"status"`
DecisionNonce string `json:"decision_nonce"`
DecisionDigest string `json:"decision_digest,omitempty"`
Revision int64 `json:"revision"`
ExpiresAt interface{} `json:"expires_at"`
Items []HTTPItemView `json:"items,omitempty"`
}
type HTTPItemView struct {
DocID string `json:"doc_id"`
Title string `json:"title,omitempty"`
ItemKey string `json:"item_key,omitempty"`
ImpactType string `json:"impact_type,omitempty"`
Depth int `json:"depth,omitempty"`
Reason string `json:"reason,omitempty"`
AuthorClaimedSnapshot bool `json:"author_claimed_snapshot"`
BusySnapshot bool `json:"busy_snapshot"`
ChangeSummary string `json:"change_summary,omitempty"`
Selected bool `json:"selected"`
Outcome string `json:"outcome,omitempty"`
ProposalID string `json:"proposal_id,omitempty"`
}
// HTTPDecision is the CAS decision envelope from HTTP.
type HTTPDecision struct {
Kind string
SelectedDocIDs []string
DecisionNonce string
DecisionDigest string
ExpectedRevision int64
}
// HTTPPort adapts *Service to the server SupplementImpactService port.
type HTTPPort struct{ Svc *Service }
func (p HTTPPort) GetBatch(ctx context.Context, batchID string) (*HTTPBatchView, error) {
if p.Svc == nil || p.Svc.Store == nil {
return nil, fmt.Errorf("supplementimpact: not configured")
}
b, err := p.Svc.Store.GetBatch(ctx, batchID)
if err != nil {
return nil, err
}
return toHTTPView(b), nil
}
func (p HTTPPort) Decide(ctx context.Context, batchID string, actorUserID int64, in HTTPDecision) (*HTTPBatchView, error) {
if p.Svc == nil {
return nil, fmt.Errorf("supplementimpact: not configured")
}
b, err := p.Svc.Decide(ctx, DecisionCommand{
BatchID: batchID, ActorUserID: actorUserID, Kind: in.Kind,
SelectedDocIDs: in.SelectedDocIDs, DecisionNonce: in.DecisionNonce,
DecisionDigest: in.DecisionDigest, ExpectedRevision: in.ExpectedRevision,
})
if err != nil {
return nil, err
}
return toHTTPView(b), nil
}
func toHTTPView(b *Batch) *HTTPBatchView {
if b == nil {
return nil
}
v := &HTTPBatchView{
BatchID: b.BatchID, ProjectID: b.ProjectID, SessionID: b.SessionID,
InitiatorUserID: b.InitiatorUserID, SourceKind: b.SourceKind, SourceSummary: b.SourceSummary,
CoverageStatus: b.CoverageStatus, Status: b.Status, DecisionNonce: b.DecisionNonce,
DecisionDigest: b.DecisionDigest, Revision: b.Revision, ExpiresAt: b.ExpiresAt,
}
for _, it := range b.Items {
v.Items = append(v.Items, HTTPItemView{
DocID: it.DocID, Title: it.Title, ItemKey: it.ItemKey, ImpactType: it.ImpactType,
Depth: it.Depth, Reason: it.Reason, AuthorClaimedSnapshot: it.AuthorClaimedSnapshot,
BusySnapshot: it.BusySnapshot, ChangeSummary: it.ChangeSummary, Selected: it.Selected,
Outcome: it.Outcome, ProposalID: it.ProposalID,
})
}
return v
}
// BuildConfirmCardJSON returns a lightweight map for coordinator InteractiveCardPayload.ProposedContent.
func BuildConfirmCardMeta(b *Batch) (title, summary string) {
if b == nil {
return "补充影响确认", ""
}
n := len(b.Items)
switch b.CoverageStatus {
case "index_pending", "partial", "failed":
return "补充影响分析未完成", fmt.Sprintf("覆盖状态:%s(不得视为无影响)。批次 %s", b.CoverageStatus, b.BatchID)
}
if n == 0 {
return "补充已记录", fmt.Sprintf("暂无文档受影响。批次 %s", b.BatchID)
}
return fmt.Sprintf("本次补充可能影响 %d 份文档", n), fmt.Sprintf("请勾选要更新的文档。批次 %s;nonce=%s;rev=%d", b.BatchID, b.DecisionNonce, b.Revision)
}
+74
View File
@@ -0,0 +1,74 @@
package supplementimpact
import (
"context"
"fmt"
"sync"
)
type MemStore struct {
mu sync.Mutex
batches map[string]*Batch
}
func NewMemStore() *MemStore {
return &MemStore{batches: map[string]*Batch{}}
}
func (m *MemStore) InsertBatch(_ context.Context, b *Batch) error {
m.mu.Lock()
defer m.mu.Unlock()
cp := *b
cp.Items = append([]Item{}, b.Items...)
m.batches[b.BatchID] = &cp
return nil
}
func (m *MemStore) GetBatch(_ context.Context, batchID string) (*Batch, error) {
m.mu.Lock()
defer m.mu.Unlock()
b, ok := m.batches[batchID]
if !ok {
return nil, fmt.Errorf("supplementimpact: batch not found")
}
cp := *b
cp.Items = append([]Item{}, b.Items...)
return &cp, nil
}
func (m *MemStore) SaveDecision(_ context.Context, batchID string, revision int64, nonce, digest, status string, selected map[string]bool) error {
m.mu.Lock()
defer m.mu.Unlock()
b, ok := m.batches[batchID]
if !ok {
return fmt.Errorf("supplementimpact: batch not found")
}
b.Status = status
b.DecisionNonce = nonce
b.DecisionDigest = digest
b.Revision = revision
for i := range b.Items {
if selected != nil {
b.Items[i].Selected = selected[b.Items[i].DocID]
}
}
return nil
}
func (m *MemStore) UpdateItemOutcome(_ context.Context, batchID, docID, outcome, proposalID, runID string) error {
m.mu.Lock()
defer m.mu.Unlock()
b, ok := m.batches[batchID]
if !ok {
return fmt.Errorf("supplementimpact: batch not found")
}
for i := range b.Items {
if b.Items[i].DocID == docID {
b.Items[i].Outcome = outcome
b.Items[i].ProposalID = proposalID
b.Items[i].RunID = runID
return nil
}
}
return fmt.Errorf("supplementimpact: item not found")
}
+173
View File
@@ -0,0 +1,173 @@
package supplementimpact
import (
"context"
"fmt"
"github.com/jackc/pgx/v5/pgxpool"
)
// PGStore persists supplement-impact batches in PostgreSQL (migration 00070).
type PGStore struct {
pool *pgxpool.Pool
}
// NewPGStore builds a Store backed by pool.
func NewPGStore(pool *pgxpool.Pool) *PGStore {
return &PGStore{pool: pool}
}
func (p *PGStore) InsertBatch(ctx context.Context, b *Batch) error {
if p == nil || p.pool == nil {
return fmt.Errorf("supplementimpact: pg store not available")
}
if b == nil {
return fmt.Errorf("supplementimpact: nil batch")
}
tx, err := p.pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
_, err = tx.Exec(ctx, `
INSERT INTO supplement_impact_batches (
batch_id, project_id, session_id, initiator_user_id, source_kind, source_summary,
coverage_status, status, decision_nonce, decision_digest, revision, expires_at
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)
`, b.BatchID, b.ProjectID, b.SessionID, b.InitiatorUserID, b.SourceKind, b.SourceSummary,
b.CoverageStatus, b.Status, b.DecisionNonce, b.DecisionDigest, b.Revision, b.ExpiresAt)
if err != nil {
return err
}
for i := range b.Items {
it := &b.Items[i]
err = tx.QueryRow(ctx, `
INSERT INTO supplement_impact_items (
batch_id, doc_id, title, item_key, impact_type, depth, reason,
author_claimed_snapshot, author_id_snapshot, busy_snapshot, change_summary,
selected, outcome, proposal_id, run_id
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15)
RETURNING id
`, it.BatchID, it.DocID, it.Title, it.ItemKey, it.ImpactType, it.Depth, it.Reason,
it.AuthorClaimedSnapshot, it.AuthorIDSnapshot, it.BusySnapshot, it.ChangeSummary,
it.Selected, it.Outcome, it.ProposalID, it.RunID).Scan(&it.ID)
if err != nil {
return err
}
}
return tx.Commit(ctx)
}
func (p *PGStore) GetBatch(ctx context.Context, batchID string) (*Batch, error) {
if p == nil || p.pool == nil {
return nil, fmt.Errorf("supplementimpact: pg store not available")
}
var b Batch
err := p.pool.QueryRow(ctx, `
SELECT batch_id, project_id, session_id, initiator_user_id, source_kind, source_summary,
coverage_status, status, decision_nonce, decision_digest, revision, expires_at
FROM supplement_impact_batches WHERE batch_id = $1
`, batchID).Scan(
&b.BatchID, &b.ProjectID, &b.SessionID, &b.InitiatorUserID, &b.SourceKind, &b.SourceSummary,
&b.CoverageStatus, &b.Status, &b.DecisionNonce, &b.DecisionDigest, &b.Revision, &b.ExpiresAt,
)
if err != nil {
return nil, fmt.Errorf("supplementimpact: batch not found: %w", err)
}
rows, err := p.pool.Query(ctx, `
SELECT id, batch_id, doc_id, title, item_key, impact_type, depth, reason,
author_claimed_snapshot, author_id_snapshot, busy_snapshot, change_summary,
selected, outcome, proposal_id, run_id
FROM supplement_impact_items WHERE batch_id = $1 ORDER BY id
`, batchID)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var it Item
if err := rows.Scan(
&it.ID, &it.BatchID, &it.DocID, &it.Title, &it.ItemKey, &it.ImpactType, &it.Depth, &it.Reason,
&it.AuthorClaimedSnapshot, &it.AuthorIDSnapshot, &it.BusySnapshot, &it.ChangeSummary,
&it.Selected, &it.Outcome, &it.ProposalID, &it.RunID,
); err != nil {
return nil, err
}
b.Items = append(b.Items, it)
}
return &b, rows.Err()
}
func (p *PGStore) SaveDecision(ctx context.Context, batchID string, revision int64, nonce, digest, status string, selected map[string]bool) error {
if p == nil || p.pool == nil {
return fmt.Errorf("supplementimpact: pg store not available")
}
tx, err := p.pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
tag, err := tx.Exec(ctx, `
UPDATE supplement_impact_batches
SET status = $1, decision_nonce = $2, decision_digest = $3, revision = $4, updated_at = now()
WHERE batch_id = $5
`, status, nonce, digest, revision, batchID)
if err != nil {
return err
}
if tag.RowsAffected() == 0 {
return fmt.Errorf("supplementimpact: batch not found")
}
if selected != nil {
rows, qerr := tx.Query(ctx, `SELECT id, doc_id FROM supplement_impact_items WHERE batch_id = $1`, batchID)
if qerr != nil {
return qerr
}
type pair struct {
id int64
docID string
}
var pairs []pair
for rows.Next() {
var pr pair
if err := rows.Scan(&pr.id, &pr.docID); err != nil {
rows.Close()
return err
}
pairs = append(pairs, pr)
}
rows.Close()
if err := rows.Err(); err != nil {
return err
}
for _, pr := range pairs {
_, err = tx.Exec(ctx, `
UPDATE supplement_impact_items SET selected = $1, updated_at = now() WHERE id = $2
`, selected[pr.docID], pr.id)
if err != nil {
return err
}
}
}
return tx.Commit(ctx)
}
func (p *PGStore) UpdateItemOutcome(ctx context.Context, batchID, docID, outcome, proposalID, runID string) error {
if p == nil || p.pool == nil {
return fmt.Errorf("supplementimpact: pg store not available")
}
tag, err := p.pool.Exec(ctx, `
UPDATE supplement_impact_items
SET outcome = $1, proposal_id = $2, run_id = $3, updated_at = now()
WHERE batch_id = $4 AND doc_id = $5
`, outcome, proposalID, runID, batchID, docID)
if err != nil {
return err
}
if tag.RowsAffected() == 0 {
return fmt.Errorf("supplementimpact: item not found")
}
return nil
}
+238
View File
@@ -0,0 +1,238 @@
package supplementimpact
import (
"context"
"crypto/rand"
"encoding/hex"
"fmt"
"strings"
"time"
"chat-agent-system/internal/docgov"
"chat-agent-system/internal/notification"
"chat-agent-system/internal/traceimpact"
)
// DocLookup resolves document claim/lease snapshots.
type DocLookup interface {
GetDocument(ctx context.Context, docID string) (*docgov.DocumentRecord, error)
}
// Analyzer runs traceimpact analysis.
type Analyzer interface {
Analyze(ctx context.Context, projectID int64, source traceimpact.DocRef, others []traceimpact.DocRef) (*traceimpact.Report, error)
}
// Store persists batches.
type Store interface {
InsertBatch(ctx context.Context, b *Batch) error
GetBatch(ctx context.Context, batchID string) (*Batch, error)
SaveDecision(ctx context.Context, batchID string, revision int64, nonce, digest, status string, selected map[string]bool) error
UpdateItemOutcome(ctx context.Context, batchID, docID, outcome, proposalID, runID string) error
}
// Notifier emits claim-aware digests.
type Notifier interface {
UpsertDocumentDigest(ctx context.Context, p notification.UpsertDigestParams) error
}
// Proposer creates docgov change proposals for claimed docs.
type Proposer interface {
ProposeModification(ctx context.Context, docID, triggeredBy, changes, impact, risk string) (*docgov.ChangeProposal, error)
}
// Service orchestrates supplement → impact → confirm → split execution.
type Service struct {
Store Store
Docs DocLookup
Analyze Analyzer
Notify Notifier
Propose Proposer
Now func() time.Time
}
func (s *Service) now() time.Time {
if s.Now != nil {
return s.Now()
}
return time.Now().UTC()
}
// CreateFromSupplement runs impact analysis and opens a confirmation batch.
func (s *Service) GetBatch(ctx context.Context, batchID string) (*Batch, error) {
if s == nil || s.Store == nil {
return nil, fmt.Errorf("supplementimpact: not configured")
}
return s.Store.GetBatch(ctx, batchID)
}
func (s *Service) CreateFromSupplement(ctx context.Context, projectID, userID int64, sessionID, summary, markdown string) (*Batch, error) {
if s == nil || s.Store == nil || s.Analyze == nil {
return nil, fmt.Errorf("supplementimpact: not configured")
}
report, err := s.Analyze.Analyze(ctx, projectID, traceimpact.DocRef{DocID: "supplement", Markdown: markdown}, nil)
if err != nil {
return nil, err
}
coverage := "no_known_dependencies"
if report != nil && report.CoverageStatus != "" {
coverage = string(report.CoverageStatus)
}
// Never present index_pending/partial/failed as "no impact".
switch coverage {
case "index_pending", "partial", "failed":
// keep as-is for the card
default:
}
batchID := "sib_" + randomHex(8)
nonce := randomHex(16)
now := s.now()
b := &Batch{
BatchID: batchID, ProjectID: projectID, SessionID: sessionID,
InitiatorUserID: userID, SourceKind: "fact_supplement",
SourceSummary: truncate(summary, 120), CoverageStatus: coverage,
Status: StatusPending, DecisionNonce: nonce, Revision: 1,
ExpiresAt: now.Add(BatchTTL),
}
seen := map[string]struct{}{}
if report != nil {
for _, imp := range report.Impacts {
if imp.DocID == "" {
continue
}
if _, ok := seen[imp.DocID]; ok {
continue
}
seen[imp.DocID] = struct{}{}
item := Item{
BatchID: batchID, DocID: imp.DocID, ItemKey: imp.ItemKey,
ImpactType: string(imp.ImpactType), Depth: imp.Depth, Reason: imp.Reason,
ChangeSummary: truncate(imp.Reason, 80),
}
if s.Docs != nil {
if doc, err := s.Docs.GetDocument(ctx, imp.DocID); err == nil && doc != nil {
item.Title = doc.Title
item.AuthorClaimedSnapshot = doc.AuthorClaimed
item.AuthorIDSnapshot = doc.AuthorID
item.BusySnapshot = doc.HasActiveLease(now)
}
}
b.Items = append(b.Items, item)
}
}
if err := s.Store.InsertBatch(ctx, b); err != nil {
return nil, err
}
return b, nil
}
// Decide confirms or cancels a batch; on confirm, executes selected items.
func (s *Service) Decide(ctx context.Context, cmd DecisionCommand) (*Batch, error) {
b, err := s.Store.GetBatch(ctx, cmd.BatchID)
if err != nil {
return nil, err
}
if b.Status != StatusPending {
return nil, fmt.Errorf("supplementimpact: batch not pending")
}
if s.now().After(b.ExpiresAt) {
_ = s.Store.SaveDecision(ctx, b.BatchID, b.Revision, cmd.DecisionNonce, cmd.DecisionDigest, StatusExpired, nil)
return nil, fmt.Errorf("supplementimpact: batch expired")
}
if cmd.ExpectedRevision != b.Revision {
return nil, fmt.Errorf("supplementimpact: revision conflict")
}
if cmd.ActorUserID != b.InitiatorUserID {
return nil, fmt.Errorf("supplementimpact: only initiator may decide")
}
if cmd.Kind == "cancel" {
if err := s.Store.SaveDecision(ctx, b.BatchID, b.Revision, cmd.DecisionNonce, cmd.DecisionDigest, StatusCancelled, nil); err != nil {
return nil, err
}
return s.Store.GetBatch(ctx, cmd.BatchID)
}
selected := map[string]bool{}
for _, id := range cmd.SelectedDocIDs {
selected[id] = true
}
if err := s.Store.SaveDecision(ctx, b.BatchID, b.Revision, cmd.DecisionNonce, cmd.DecisionDigest, StatusRunning, selected); err != nil {
return nil, err
}
b, _ = s.Store.GetBatch(ctx, cmd.BatchID)
for i := range b.Items {
it := &b.Items[i]
if !selected[it.DocID] {
continue
}
if it.BusySnapshot {
_ = s.Store.UpdateItemOutcome(ctx, b.BatchID, it.DocID, OutcomeSkippedBusy, "", "")
continue
}
if err := s.executeItem(ctx, b, it); err != nil {
_ = s.Store.UpdateItemOutcome(ctx, b.BatchID, it.DocID, OutcomeFailed, "", "")
}
}
final := StatusDone
b, _ = s.Store.GetBatch(ctx, cmd.BatchID)
for _, it := range b.Items {
if !it.Selected {
continue
}
switch it.Outcome {
case OutcomeSkippedBusy, OutcomeSkippedLocked, OutcomeFailed, OutcomeProposalPending:
final = StatusPartiallyDone
}
}
_ = s.Store.SaveDecision(ctx, b.BatchID, b.Revision+1, cmd.DecisionNonce, cmd.DecisionDigest, final, selected)
return s.Store.GetBatch(ctx, cmd.BatchID)
}
func (s *Service) executeItem(ctx context.Context, b *Batch, it *Item) error {
if it.AuthorClaimedSnapshot {
if s.Propose == nil {
return fmt.Errorf("proposer not configured")
}
prop, err := s.Propose.ProposeModification(ctx, it.DocID, fmt.Sprintf("user:%d", b.InitiatorUserID),
it.ChangeSummary, it.Reason, "补充材料触发的已认领文档变更,需编制负责人确认")
if err != nil {
return err
}
_ = s.Store.UpdateItemOutcome(ctx, b.BatchID, it.DocID, OutcomeProposalPending, prop.ProposalID, "")
if s.Notify != nil && it.AuthorIDSnapshot > 0 {
_ = s.Notify.UpsertDocumentDigest(ctx, notification.UpsertDigestParams{
UserID: it.AuthorIDSnapshot, ProjectID: b.ProjectID,
DigestKind: notification.DigestKindProposalRequired, DocID: it.DocID, DocTitle: it.Title,
Severity: notification.SeverityActionRequired,
ActorID: b.InitiatorUserID, ActorName: fmt.Sprintf("user:%d", b.InitiatorUserID),
Summary: b.SourceSummary, ProposalID: prop.ProposalID,
})
}
return nil
}
// Unclaimed: notify info-only; actual content rewrite is owned by squad/task path.
_ = s.Store.UpdateItemOutcome(ctx, b.BatchID, it.DocID, OutcomeApplied, "", "")
if s.Notify != nil && it.AuthorIDSnapshot > 0 {
_ = s.Notify.UpsertDocumentDigest(ctx, notification.UpsertDigestParams{
UserID: it.AuthorIDSnapshot, ProjectID: b.ProjectID,
DigestKind: notification.DigestKindUnclaimedEdit, DocID: it.DocID, DocTitle: it.Title,
Severity: notification.SeverityInfo,
ActorID: b.InitiatorUserID, ActorName: fmt.Sprintf("user:%d", b.InitiatorUserID),
Summary: b.SourceSummary,
})
}
return nil
}
func randomHex(n int) string {
b := make([]byte, n)
_, _ = rand.Read(b)
return hex.EncodeToString(b)
}
func truncate(s string, n int) string {
r := []rune(strings.TrimSpace(s))
if len(r) <= n {
return string(r)
}
return string(r[:n]) + "…"
}
+108
View File
@@ -0,0 +1,108 @@
package supplementimpact_test
import (
"context"
"testing"
"time"
"chat-agent-system/internal/docgov"
"chat-agent-system/internal/notification"
"chat-agent-system/internal/supplementimpact"
"chat-agent-system/internal/traceimpact"
)
type stubAnalyzer struct{ report *traceimpact.Report }
func (s stubAnalyzer) Analyze(_ context.Context, _ int64, _ traceimpact.DocRef, _ []traceimpact.DocRef) (*traceimpact.Report, error) {
return s.report, nil
}
type stubDocs struct{ docs map[string]*docgov.DocumentRecord }
func (s stubDocs) GetDocument(_ context.Context, id string) (*docgov.DocumentRecord, error) {
return s.docs[id], nil
}
func TestCreateBatchSnapshotsClaimAndBusy(t *testing.T) {
store := supplementimpact.NewMemStore()
claimedAt := time.Now()
svc := &supplementimpact.Service{
Store: store,
Analyze: stubAnalyzer{report: &traceimpact.Report{
CoverageStatus: "confirmed_impact",
Impacts: []traceimpact.Impact{
{DocID: "doc_a", ItemKey: "UR-1", ImpactType: "confirmed_impact", Depth: 1, Reason: "hit"},
},
}},
Docs: stubDocs{docs: map[string]*docgov.DocumentRecord{
"doc_a": {DocID: "doc_a", Title: "需求", AuthorID: 9, AuthorClaimed: true, AuthorClaimedAt: &claimedAt},
}},
}
b, err := svc.CreateFromSupplement(context.Background(), 1, 2, "sess", "补充市场", "UR-1 目标市场改为欧盟")
if err != nil {
t.Fatal(err)
}
if len(b.Items) != 1 || !b.Items[0].AuthorClaimedSnapshot {
t.Fatalf("want claimed snapshot, got %+v", b.Items)
}
}
type stubPropose struct {
lastDoc string
prop *docgov.ChangeProposal
}
func (s *stubPropose) ProposeModification(_ context.Context, docID, _, _, _, _ string) (*docgov.ChangeProposal, error) {
s.lastDoc = docID
if s.prop == nil {
s.prop = &docgov.ChangeProposal{ProposalID: "prop_test_1", DocID: docID, Status: docgov.ProposalPendingAuthorApproval}
}
return s.prop, nil
}
type stubNotify struct{ n int }
func (s *stubNotify) UpsertDocumentDigest(_ context.Context, _ notification.UpsertDigestParams) error {
s.n++
return nil
}
func TestDecideClaimedProducesProposal(t *testing.T) {
store := supplementimpact.NewMemStore()
claimedAt := time.Now()
prop := &stubPropose{}
notify := &stubNotify{}
svc := &supplementimpact.Service{
Store: store,
Analyze: stubAnalyzer{report: &traceimpact.Report{
CoverageStatus: "confirmed_impact",
Impacts: []traceimpact.Impact{
{DocID: "doc_a", ItemKey: "UR-1", ImpactType: "confirmed_impact", Depth: 1, Reason: "hit"},
},
}},
Docs: stubDocs{docs: map[string]*docgov.DocumentRecord{
"doc_a": {DocID: "doc_a", Title: "需求", AuthorID: 9, AuthorClaimed: true, AuthorClaimedAt: &claimedAt},
}},
Propose: prop,
Notify: notify,
}
b, err := svc.CreateFromSupplement(context.Background(), 1, 2, "sess", "补充市场", "UR-1 目标市场改为欧盟")
if err != nil {
t.Fatal(err)
}
out, err := svc.Decide(context.Background(), supplementimpact.DecisionCommand{
BatchID: b.BatchID, ActorUserID: 2, Kind: "confirm",
SelectedDocIDs: []string{"doc_a"}, DecisionNonce: b.DecisionNonce,
DecisionDigest: "d1", ExpectedRevision: b.Revision,
})
if err != nil {
t.Fatal(err)
}
if len(out.Items) != 1 || out.Items[0].Outcome != supplementimpact.OutcomeProposalPending {
t.Fatalf("want proposal_pending, got %+v", out.Items)
}
if prop.lastDoc != "doc_a" || notify.n != 1 {
t.Fatalf("propose/notify not called: doc=%s notify=%d", prop.lastDoc, notify.n)
}
}
+28
View File
@@ -0,0 +1,28 @@
package supplementimpact
import (
"context"
"log/slog"
"time"
)
// TriggerAsync kicks CreateFromSupplement in a background goroutine with panic recovery
// so chat / mutate paths are never blocked by impact analysis.
func TriggerAsync(svc *Service, projectID, userID int64, sessionID, summary, markdown string) {
if svc == nil {
return
}
go func() {
defer func() {
if rec := recover(); rec != nil {
slog.Error("supplementimpact: TriggerAsync panic", "recover", rec, "project_id", projectID)
}
}()
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
if _, err := svc.CreateFromSupplement(ctx, projectID, userID, sessionID, summary, markdown); err != nil {
slog.Warn("supplementimpact: CreateFromSupplement failed",
"project_id", projectID, "user_id", userID, "error", err)
}
}()
}
+70
View File
@@ -0,0 +1,70 @@
package supplementimpact
import "time"
const (
StatusPending = "pending"
StatusDecided = "decided"
StatusRunning = "running"
StatusDone = "done"
StatusPartiallyDone = "partially_done"
StatusExpired = "expired"
StatusCancelled = "cancelled"
OutcomeApplied = "applied"
OutcomeProposalPending = "proposal_pending"
OutcomeProposalRejected = "proposal_rejected"
OutcomeSkippedBusy = "skipped_busy"
OutcomeSkippedLocked = "skipped_locked"
OutcomeFailed = "failed"
BatchTTL = 24 * time.Hour
)
// Batch is one supplement → impact analysis confirmation unit.
type Batch struct {
BatchID string `json:"batch_id"`
ProjectID int64 `json:"project_id"`
SessionID string `json:"session_id"`
InitiatorUserID int64 `json:"initiator_user_id"`
SourceKind string `json:"source_kind"`
SourceSummary string `json:"source_summary"`
CoverageStatus string `json:"coverage_status"`
Status string `json:"status"`
DecisionNonce string `json:"decision_nonce"`
DecisionDigest string `json:"decision_digest"`
Revision int64 `json:"revision"`
ExpiresAt time.Time `json:"expires_at"`
Items []Item `json:"items,omitempty"`
}
// Item is one affected document inside a batch.
type Item struct {
ID int64 `json:"id"`
BatchID string `json:"batch_id"`
DocID string `json:"doc_id"`
Title string `json:"title"`
ItemKey string `json:"item_key"`
ImpactType string `json:"impact_type"`
Depth int `json:"depth"`
Reason string `json:"reason"`
AuthorClaimedSnapshot bool `json:"author_claimed_snapshot"`
AuthorIDSnapshot int64 `json:"author_id_snapshot"`
BusySnapshot bool `json:"busy_snapshot"`
ChangeSummary string `json:"change_summary"`
Selected bool `json:"selected"`
Outcome string `json:"outcome"`
ProposalID string `json:"proposal_id"`
RunID string `json:"run_id"`
}
// DecisionCommand is the confirm/cancel envelope (nonce+digest+revision CAS).
type DecisionCommand struct {
BatchID string
ActorUserID int64
Kind string // confirm | cancel
SelectedDocIDs []string
DecisionNonce string
DecisionDigest string
ExpectedRevision int64
}