From 4e0e6a1418840f4040925e5acc3a6c8fc200d62a Mon Sep 17 00:00:00 2001 From: weichen Date: Thu, 24 Sep 2026 11:44:20 +0000 Subject: [PATCH] feat(notification): digest aggregates, supplement-impact batch, and free-collab hooks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../2026-09-24-free-collab-claim-notify.md | 22 ++ ...6-09-24-free-collab-claim-notify-design.md | 58 +++++ .doc/system-design/README.md | 9 + .doc/system-design/module-boundaries.yaml | 26 ++ ...026-09-24-notification-digest-severity.sql | 16 ++ .../00069_2026-09-24-notification-digests.sql | 25 ++ ...0_2026-09-24-supplement-impact-batches.sql | 49 ++++ internal/app/app.go | 7 +- internal/app/docnotify/consumer.go | 4 + internal/app/mutate_owner.go | 17 ++ internal/app/wire_supplement_impact.go | 72 ++++++ internal/bizdev/after_proposal.go | 50 ++++ internal/bizdev/after_proposal_test.go | 63 +++++ internal/bizdev/service.go | 16 ++ internal/bizdev/service_task_run.go | 15 ++ internal/coordinator/engine.go | 27 ++ internal/coordinator/engine_stage.go | 15 ++ internal/events/registry.go | 4 + internal/notification/digest.go | 186 ++++++++++++++ internal/notification/document_events.go | 4 + internal/notification/service.go | 52 +++- internal/notification/types.go | 23 +- internal/supplementimpact/http_adapter.go | 113 +++++++++ internal/supplementimpact/mem_store.go | 74 ++++++ internal/supplementimpact/pg_store.go | 173 +++++++++++++ internal/supplementimpact/service.go | 238 ++++++++++++++++++ internal/supplementimpact/service_test.go | 108 ++++++++ internal/supplementimpact/trigger.go | 28 +++ internal/supplementimpact/types.go | 70 ++++++ 29 files changed, 1546 insertions(+), 18 deletions(-) create mode 100644 .doc/ops/short-term/2026-09-24-free-collab-claim-notify.md create mode 100644 .doc/system-design/2026-09-24-free-collab-claim-notify-design.md create mode 100644 deploy/postgres/goose/00068_2026-09-24-notification-digest-severity.sql create mode 100644 deploy/postgres/goose/00069_2026-09-24-notification-digests.sql create mode 100644 deploy/postgres/goose/00070_2026-09-24-supplement-impact-batches.sql create mode 100644 internal/app/wire_supplement_impact.go create mode 100644 internal/bizdev/after_proposal.go create mode 100644 internal/bizdev/after_proposal_test.go create mode 100644 internal/notification/digest.go create mode 100644 internal/supplementimpact/http_adapter.go create mode 100644 internal/supplementimpact/mem_store.go create mode 100644 internal/supplementimpact/pg_store.go create mode 100644 internal/supplementimpact/service.go create mode 100644 internal/supplementimpact/service_test.go create mode 100644 internal/supplementimpact/trigger.go create mode 100644 internal/supplementimpact/types.go diff --git a/.doc/ops/short-term/2026-09-24-free-collab-claim-notify.md b/.doc/ops/short-term/2026-09-24-free-collab-claim-notify.md new file mode 100644 index 00000000..a610c325 --- /dev/null +++ b/.doc/ops/short-term/2026-09-24-free-collab-claim-notify.md @@ -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 补充影响批次 + 确认卡 + 分流执行 diff --git a/.doc/system-design/2026-09-24-free-collab-claim-notify-design.md b/.doc/system-design/2026-09-24-free-collab-claim-notify-design.md new file mode 100644 index 00000000..77ff4e1b --- /dev/null +++ b/.doc/system-design/2026-09-24-free-collab-claim-notify-design.md @@ -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 表 +- 索引缺失不得显示为无影响 diff --git a/.doc/system-design/README.md b/.doc/system-design/README.md index b91a5fbe..25e7ce4b 100644 --- a/.doc/system-design/README.md +++ b/.doc/system-design/README.md @@ -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) | 项 | 链接 | diff --git a/.doc/system-design/module-boundaries.yaml b/.doc/system-design/module-boundaries.yaml index 700549a9..c3711a3d 100644 --- a/.doc/system-design/module-boundaries.yaml +++ b/.doc/system-design/module-boundaries.yaml @@ -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 diff --git a/deploy/postgres/goose/00068_2026-09-24-notification-digest-severity.sql b/deploy/postgres/goose/00068_2026-09-24-notification-digest-severity.sql new file mode 100644 index 00000000..fdaedb88 --- /dev/null +++ b/deploy/postgres/goose/00068_2026-09-24-notification-digest-severity.sql @@ -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; diff --git a/deploy/postgres/goose/00069_2026-09-24-notification-digests.sql b/deploy/postgres/goose/00069_2026-09-24-notification-digests.sql new file mode 100644 index 00000000..f4851e7f --- /dev/null +++ b/deploy/postgres/goose/00069_2026-09-24-notification-digests.sql @@ -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; diff --git a/deploy/postgres/goose/00070_2026-09-24-supplement-impact-batches.sql b/deploy/postgres/goose/00070_2026-09-24-supplement-impact-batches.sql new file mode 100644 index 00000000..dfffedf0 --- /dev/null +++ b/deploy/postgres/goose/00070_2026-09-24-supplement-impact-batches.sql @@ -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; diff --git a/internal/app/app.go b/internal/app/app.go index 795c4dc9..f855d855 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -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 } diff --git a/internal/app/docnotify/consumer.go b/internal/app/docnotify/consumer.go index 8d70d6d8..ee3422c0 100644 --- a/internal/app/docnotify/consumer.go +++ b/internal/app/docnotify/consumer.go @@ -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 "", "", "" } diff --git a/internal/app/mutate_owner.go b/internal/app/mutate_owner.go index 378ac8b8..837df2e1 100644 --- a/internal/app/mutate_owner.go +++ b/internal/app/mutate_owner.go @@ -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{ diff --git a/internal/app/wire_supplement_impact.go b/internal/app/wire_supplement_impact.go new file mode 100644 index 00000000..9f09ca79 --- /dev/null +++ b/internal/app/wire_supplement_impact.go @@ -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, + } +} diff --git a/internal/bizdev/after_proposal.go b/internal/bizdev/after_proposal.go new file mode 100644 index 00000000..fb587d1e --- /dev/null +++ b/internal/bizdev/after_proposal.go @@ -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) +} diff --git a/internal/bizdev/after_proposal_test.go b/internal/bizdev/after_proposal_test.go new file mode 100644 index 00000000..fde9fb14 --- /dev/null +++ b/internal/bizdev/after_proposal_test.go @@ -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") + } +} diff --git a/internal/bizdev/service.go b/internal/bizdev/service.go index 177e4812..823aa688 100644 --- a/internal/bizdev/service.go +++ b/internal/bizdev/service.go @@ -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 { diff --git a/internal/bizdev/service_task_run.go b/internal/bizdev/service_task_run.go index 375f5683..3dff8e02 100644 --- a/internal/bizdev/service_task_run.go +++ b/internal/bizdev/service_task_run.go @@ -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{ diff --git a/internal/coordinator/engine.go b/internal/coordinator/engine.go index c4c09ef0..7d853fb4 100644 --- a/internal/coordinator/engine.go +++ b/internal/coordinator/engine.go @@ -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, diff --git a/internal/coordinator/engine_stage.go b/internal/coordinator/engine_stage.go index 320765ce..281c760e 100644 --- a/internal/coordinator/engine_stage.go +++ b/internal/coordinator/engine_stage.go @@ -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 } diff --git a/internal/events/registry.go b/internal/events/registry.go index f615123f..4b96f879 100644 --- a/internal/events/registry.go +++ b/internal/events/registry.go @@ -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"}}, } } diff --git a/internal/notification/digest.go b/internal/notification/digest.go new file mode 100644 index 00000000..1f6b2d2f --- /dev/null +++ b/internal/notification/digest.go @@ -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, ¬ifID) + 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(¬ifID) + 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. diff --git a/internal/notification/document_events.go b/internal/notification/document_events.go index d22cffe7..0b0ef62d 100644 --- a/internal/notification/document_events.go +++ b/internal/notification/document_events.go @@ -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). diff --git a/internal/notification/service.go b/internal/notification/service.go index 368d358e..8885b1f0 100644 --- a/internal/notification/service.go +++ b/internal/notification/service.go @@ -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() diff --git a/internal/notification/types.go b/internal/notification/types.go index 6c829e83..1b0e0ae9 100644 --- a/internal/notification/types.go +++ b/internal/notification/types.go @@ -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"` } diff --git a/internal/supplementimpact/http_adapter.go b/internal/supplementimpact/http_adapter.go new file mode 100644 index 00000000..b34e7d38 --- /dev/null +++ b/internal/supplementimpact/http_adapter.go @@ -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) +} diff --git a/internal/supplementimpact/mem_store.go b/internal/supplementimpact/mem_store.go new file mode 100644 index 00000000..7f3901c6 --- /dev/null +++ b/internal/supplementimpact/mem_store.go @@ -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") +} diff --git a/internal/supplementimpact/pg_store.go b/internal/supplementimpact/pg_store.go new file mode 100644 index 00000000..fa78debc --- /dev/null +++ b/internal/supplementimpact/pg_store.go @@ -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 +} diff --git a/internal/supplementimpact/service.go b/internal/supplementimpact/service.go new file mode 100644 index 00000000..bc6a110f --- /dev/null +++ b/internal/supplementimpact/service.go @@ -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]) + "…" +} diff --git a/internal/supplementimpact/service_test.go b/internal/supplementimpact/service_test.go new file mode 100644 index 00000000..45c42ed2 --- /dev/null +++ b/internal/supplementimpact/service_test.go @@ -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) + } +} + diff --git a/internal/supplementimpact/trigger.go b/internal/supplementimpact/trigger.go new file mode 100644 index 00000000..167e4ca0 --- /dev/null +++ b/internal/supplementimpact/trigger.go @@ -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) + } + }() +} diff --git a/internal/supplementimpact/types.go b/internal/supplementimpact/types.go new file mode 100644 index 00000000..3e031820 --- /dev/null +++ b/internal/supplementimpact/types.go @@ -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 +}