From 33ee59641f97fc19041cf8c155f7fdf3db6fdf80 Mon Sep 17 00:00:00 2001 From: weichen Date: Sun, 20 Sep 2026 15:35:07 +0000 Subject: [PATCH] feat(trace): add the project trace ledger and bounded impact walk Keep item revisions and edge decisions as records, and refuse to treat a missing index as no impact. Co-authored-by: Cursor --- .../goose/00059_2026-09-20-project-trace.sql | 108 +++++ internal/traceimpact/extract.go | 74 ++++ internal/traceimpact/migration_test.go | 27 ++ internal/traceimpact/open.go | 28 ++ internal/traceimpact/pg_store.go | 307 ++++++++++++++ internal/traceimpact/policy.go | 114 ++++++ internal/traceimpact/service.go | 380 ++++++++++++++++++ internal/traceimpact/service_test.go | 230 +++++++++++ internal/traceimpact/store.go | 317 +++++++++++++++ internal/traceimpact/types.go | 164 ++++++++ internal/traceimpact/walk.go | 59 +++ 11 files changed, 1808 insertions(+) create mode 100644 deploy/postgres/goose/00059_2026-09-20-project-trace.sql create mode 100644 internal/traceimpact/extract.go create mode 100644 internal/traceimpact/migration_test.go create mode 100644 internal/traceimpact/open.go create mode 100644 internal/traceimpact/pg_store.go create mode 100644 internal/traceimpact/policy.go create mode 100644 internal/traceimpact/service.go create mode 100644 internal/traceimpact/service_test.go create mode 100644 internal/traceimpact/store.go create mode 100644 internal/traceimpact/types.go create mode 100644 internal/traceimpact/walk.go diff --git a/deploy/postgres/goose/00059_2026-09-20-project-trace.sql b/deploy/postgres/goose/00059_2026-09-20-project-trace.sql new file mode 100644 index 00000000..6d020da4 --- /dev/null +++ b/deploy/postgres/goose/00059_2026-09-20-project-trace.sql @@ -0,0 +1,108 @@ +-- +goose Up +-- 项目追踪与阶段回溯账本。留存类别为 record：随项目保留。 +-- 条目修订与边决议只追加。源文档依法清除时把条目标成 tombstone，不级联删边和决议。 + +CREATE TABLE IF NOT EXISTS project_trace_items ( + id BIGSERIAL PRIMARY KEY, + project_id BIGINT NOT NULL, + business_key TEXT NOT NULL, + kind TEXT NOT NULL, + lifecycle_state TEXT NOT NULL DEFAULT 'active', + review_state TEXT NOT NULL DEFAULT 'confirmed', + tombstone BOOLEAN NOT NULL DEFAULT false, + statement TEXT NOT NULL DEFAULT '', + source_doc_id TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (project_id, kind, business_key) +); + +CREATE TABLE IF NOT EXISTS project_trace_item_revisions ( + id BIGSERIAL PRIMARY KEY, + item_id BIGINT NOT NULL REFERENCES project_trace_items (id), + revision_no INT NOT NULL, + source_doc_id TEXT NOT NULL DEFAULT '', + source_snapshot_id BIGINT NOT NULL DEFAULT 0, + content_hash TEXT NOT NULL DEFAULT '', + statement TEXT NOT NULL DEFAULT '', + extractor_version TEXT NOT NULL DEFAULT '', + tombstone BOOLEAN NOT NULL DEFAULT false, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (item_id, revision_no), + UNIQUE (item_id, source_snapshot_id, content_hash, extractor_version) +); + +CREATE TABLE IF NOT EXISTS project_trace_edges ( + id BIGSERIAL PRIMARY KEY, + project_id BIGINT NOT NULL, + from_item_id BIGINT NOT NULL, + to_item_id BIGINT NOT NULL, + relation_type TEXT NOT NULL, + provenance TEXT NOT NULL, + review_state TEXT NOT NULL, + evidence_doc_id TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (project_id, from_item_id, to_item_id, relation_type, provenance) +); + +CREATE TABLE IF NOT EXISTS project_trace_edge_decisions ( + id BIGSERIAL PRIMARY KEY, + project_id BIGINT NOT NULL, + edge_id BIGINT NOT NULL, + decision TEXT NOT NULL, + reason TEXT NOT NULL DEFAULT '', + decided_by BIGINT NOT NULL, + decided_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS project_trace_index_state ( + project_id BIGINT NOT NULL, + doc_id TEXT NOT NULL, + source_snapshot_id BIGINT NOT NULL DEFAULT 0, + status TEXT NOT NULL, + extractor_version TEXT NOT NULL, + item_count INT NOT NULL DEFAULT 0, + error TEXT NOT NULL DEFAULT '', + indexed_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (project_id, doc_id, extractor_version) +); + +CREATE TABLE IF NOT EXISTS stage_reopen_requests ( + id BIGSERIAL PRIMARY KEY, + project_id BIGINT NOT NULL, + instance_id BIGINT NOT NULL, + stage_key TEXT NOT NULL, + doc_task_keys TEXT[] NOT NULL, + reason TEXT NOT NULL, + requested_by BIGINT NOT NULL, + status TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS stage_recheck_marks ( + id BIGSERIAL PRIMARY KEY, + project_id BIGINT NOT NULL, + stage_key TEXT NOT NULL, + source_doc_id TEXT NOT NULL, + status TEXT NOT NULL, + reason TEXT NOT NULL DEFAULT '', + acked_by BIGINT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE UNIQUE INDEX IF NOT EXISTS stage_recheck_marks_one_open + ON stage_recheck_marks (project_id, stage_key, source_doc_id) + WHERE status = 'open'; + +CREATE TABLE IF NOT EXISTS trace_revision_proposals ( + id BIGSERIAL PRIMARY KEY, + project_id BIGINT NOT NULL, + author_id BIGINT NOT NULL, + body TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending_author', + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- +goose Down +-- 这些表是合规业务记录，不是可丢弃缓存。 +-- Down 不删除 project_trace_edge_decisions，也不删除条目、修订、边、索引水位、再打开申请和待复核标记。 +SELECT 1; diff --git a/internal/traceimpact/extract.go b/internal/traceimpact/extract.go new file mode 100644 index 00000000..f9b5d94b --- /dev/null +++ b/internal/traceimpact/extract.go @@ -0,0 +1,74 @@ +package traceimpact + +import ( + "regexp" + "strings" +) + +var ( + explicitKeyRe = regexp.MustCompile(`\b((?:UR|RS|REQ|TC|RF|RP|DI)-\d+)\b`) + untypedKeyRe = regexp.MustCompile(`\b([A-Z]{2,10}-\d+)\b`) +) + +// ExtractedKey is a business key found in markdown. +type ExtractedKey struct { + Key string + Kind string + Explicit bool + Line string +} + +// ExtractKeys returns explicit UR/RS/REQ/TC-style keys and untyped ID mentions. +// Untyped mentions stay candidates and never become confirmed edges by themselves. +func ExtractKeys(markdown string) []ExtractedKey { + lines := strings.Split(markdown, "\n") + var out []ExtractedKey + seen := map[string]bool{} + for _, line := range lines { + for _, m := range explicitKeyRe.FindAllStringSubmatch(line, -1) { + key := m[1] + if seen["e:"+key] { + continue + } + seen["e:"+key] = true + kind := KindRequirement + if strings.HasPrefix(key, "TC-") { + kind = KindTest + } + out = append(out, ExtractedKey{Key: key, Kind: kind, Explicit: true, Line: line}) + } + for _, m := range untypedKeyRe.FindAllStringSubmatch(line, -1) { + key := m[1] + if explicitKeyRe.MatchString(key) || seen["u:"+key] || seen["e:"+key] { + continue + } + seen["u:"+key] = true + out = append(out, ExtractedKey{Key: key, Kind: KindRequirement, Explicit: false, Line: line}) + } + } + return out +} + +// ExplicitPairsOnLine returns ordered pairs of explicit keys that share a line. +func ExplicitPairsOnLine(markdown string) [][2]string { + var pairs [][2]string + seen := map[string]bool{} + for _, line := range strings.Split(markdown, "\n") { + found := explicitKeyRe.FindAllStringSubmatch(line, -1) + if len(found) < 2 { + continue + } + for i := 0; i < len(found); i++ { + for j := i + 1; j < len(found); j++ { + a, b := found[i][1], found[j][1] + k := a + ">" + b + if seen[k] { + continue + } + seen[k] = true + pairs = append(pairs, [2]string{a, b}) + } + } + } + return pairs +} diff --git a/internal/traceimpact/migration_test.go b/internal/traceimpact/migration_test.go new file mode 100644 index 00000000..514ed251 --- /dev/null +++ b/internal/traceimpact/migration_test.go @@ -0,0 +1,27 @@ +package traceimpact + +import ( + "os" + "strings" + "testing" +) + +func TestMigrationDownKeepsDecisionLedger(t *testing.T) { + b, err := os.ReadFile("../../deploy/postgres/goose/00059_2026-09-20-project-trace.sql") + if err != nil { + t.Fatal(err) + } + parts := strings.Split(string(b), "-- +goose Down") + if len(parts) != 2 { + t.Fatal("missing goose down") + } + down := strings.ToLower(parts[1]) + for _, forbidden := range []string{"drop table", "delete from", "truncate"} { + if strings.Contains(down, forbidden) { + t.Fatalf("down migration must not %s the trace ledger", forbidden) + } + } + if !strings.Contains(down, "project_trace_edge_decisions") { + t.Fatal("down comment should name the decision ledger it refuses to drop") + } +} diff --git a/internal/traceimpact/open.go b/internal/traceimpact/open.go new file mode 100644 index 00000000..b76642ec --- /dev/null +++ b/internal/traceimpact/open.go @@ -0,0 +1,28 @@ +package traceimpact + +import ( + "context" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// Open uses the record tables when migration 00059 is present. +// Before that migration, it keeps an in-memory ledger so the process can still boot. +func Open(pool *pgxpool.Pool) *Service { + if pool == nil { + return NewMemoryService() + } + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + var exists bool + err := pool.QueryRow(ctx, ` + SELECT EXISTS ( + SELECT 1 FROM information_schema.tables + WHERE table_schema = 'public' AND table_name = 'project_trace_items' + )`).Scan(&exists) + if err != nil || !exists { + return NewMemoryService() + } + return NewService(newPGStore(pool)) +} diff --git a/internal/traceimpact/pg_store.go b/internal/traceimpact/pg_store.go new file mode 100644 index 00000000..3937f72c --- /dev/null +++ b/internal/traceimpact/pg_store.go @@ -0,0 +1,307 @@ +package traceimpact + +import ( + "context" + "fmt" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +type pgStore struct { + pool *pgxpool.Pool +} + +func newPGStore(pool *pgxpool.Pool) *pgStore { + return &pgStore{pool: pool} +} + +func (p *pgStore) UpsertItem(ctx context.Context, item Item) (Item, error) { + if item.Lifecycle == "" { + item.Lifecycle = "active" + } + row := p.pool.QueryRow(ctx, ` + INSERT INTO project_trace_items + (project_id, business_key, kind, lifecycle_state, review_state, tombstone, statement, source_doc_id) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8) + ON CONFLICT (project_id, kind, business_key) DO UPDATE SET + review_state = EXCLUDED.review_state, + tombstone = EXCLUDED.tombstone, + statement = CASE WHEN EXCLUDED.statement <> '' THEN EXCLUDED.statement ELSE project_trace_items.statement END, + source_doc_id = CASE WHEN EXCLUDED.source_doc_id <> '' THEN EXCLUDED.source_doc_id ELSE project_trace_items.source_doc_id END + RETURNING id, project_id, business_key, kind, lifecycle_state, review_state, tombstone, statement, source_doc_id + `, item.ProjectID, item.BusinessKey, item.Kind, item.Lifecycle, item.ReviewState, item.Tombstone, item.Statement, item.SourceDocID) + return scanItem(row) +} + +func scanItem(row pgx.Row) (Item, error) { + var item Item + err := row.Scan(&item.ID, &item.ProjectID, &item.BusinessKey, &item.Kind, &item.Lifecycle, &item.ReviewState, &item.Tombstone, &item.Statement, &item.SourceDocID) + return item, err +} + +func (p *pgStore) ListItems(ctx context.Context, projectID int64) ([]Item, error) { + rows, err := p.pool.Query(ctx, ` + SELECT id, project_id, business_key, kind, lifecycle_state, review_state, tombstone, statement, source_doc_id + FROM project_trace_items WHERE project_id = $1 ORDER BY id + `, projectID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []Item + for rows.Next() { + var item Item + if err := rows.Scan(&item.ID, &item.ProjectID, &item.BusinessKey, &item.Kind, &item.Lifecycle, &item.ReviewState, &item.Tombstone, &item.Statement, &item.SourceDocID); err != nil { + return nil, err + } + out = append(out, item) + } + return out, rows.Err() +} + +func (p *pgStore) UpsertEdge(ctx context.Context, edge Edge) (Edge, error) { + row := p.pool.QueryRow(ctx, ` + INSERT INTO project_trace_edges + (project_id, from_item_id, to_item_id, relation_type, provenance, review_state, evidence_doc_id) + VALUES ($1,$2,$3,$4,$5,$6,$7) + ON CONFLICT (project_id, from_item_id, to_item_id, relation_type, provenance) DO UPDATE SET + review_state = CASE + WHEN project_trace_edges.review_state = 'rejected' AND EXCLUDED.review_state = 'candidate' + THEN project_trace_edges.review_state + WHEN project_trace_edges.review_state = 'confirmed' + THEN project_trace_edges.review_state + ELSE EXCLUDED.review_state + END + RETURNING id, project_id, from_item_id, to_item_id, relation_type, provenance, review_state, evidence_doc_id + `, edge.ProjectID, edge.FromItemID, edge.ToItemID, edge.RelationType, edge.Provenance, edge.ReviewState, edge.EvidenceDoc) + var out Edge + err := row.Scan(&out.ID, &out.ProjectID, &out.FromItemID, &out.ToItemID, &out.RelationType, &out.Provenance, &out.ReviewState, &out.EvidenceDoc) + return out, err +} + +func (p *pgStore) ListEdges(ctx context.Context, projectID int64) ([]Edge, error) { + rows, err := p.pool.Query(ctx, ` + SELECT id, project_id, from_item_id, to_item_id, relation_type, provenance, review_state, evidence_doc_id + FROM project_trace_edges WHERE project_id = $1 ORDER BY id + `, projectID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []Edge + for rows.Next() { + var e Edge + if err := rows.Scan(&e.ID, &e.ProjectID, &e.FromItemID, &e.ToItemID, &e.RelationType, &e.Provenance, &e.ReviewState, &e.EvidenceDoc); err != nil { + return nil, err + } + out = append(out, e) + } + return out, rows.Err() +} + +func (p *pgStore) PutIndex(ctx context.Context, st IndexState) error { + if st.Extractor == "" { + st.Extractor = ExtractorVersion + } + _, err := p.pool.Exec(ctx, ` + INSERT INTO project_trace_index_state + (project_id, doc_id, source_snapshot_id, status, extractor_version, item_count, error) + VALUES ($1,$2,$3,$4,$5,$6,$7) + ON CONFLICT (project_id, doc_id, extractor_version) DO UPDATE SET + source_snapshot_id = EXCLUDED.source_snapshot_id, + status = EXCLUDED.status, + item_count = EXCLUDED.item_count, + error = EXCLUDED.error, + indexed_at = now() + `, st.ProjectID, st.DocID, st.SnapshotID, st.Status, st.Extractor, st.ItemCount, st.Error) + return err +} + +func (p *pgStore) IndexStatus(ctx context.Context, projectID int64, docID string) (IndexState, bool, error) { + row := p.pool.QueryRow(ctx, ` + SELECT project_id, doc_id, source_snapshot_id, status, extractor_version, item_count, error + FROM project_trace_index_state + WHERE project_id = $1 AND doc_id = $2 + ORDER BY indexed_at DESC LIMIT 1 + `, projectID, docID) + var st IndexState + err := row.Scan(&st.ProjectID, &st.DocID, &st.SnapshotID, &st.Status, &st.Extractor, &st.ItemCount, &st.Error) + if err == pgx.ErrNoRows { + return IndexState{}, false, nil + } + if err != nil { + return IndexState{}, false, err + } + return st, true, nil +} + +func (p *pgStore) ListIndex(ctx context.Context, projectID int64) ([]IndexState, error) { + rows, err := p.pool.Query(ctx, ` + SELECT project_id, doc_id, source_snapshot_id, status, extractor_version, item_count, error + FROM project_trace_index_state WHERE project_id = $1 ORDER BY doc_id + `, projectID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []IndexState + for rows.Next() { + var st IndexState + if err := rows.Scan(&st.ProjectID, &st.DocID, &st.SnapshotID, &st.Status, &st.Extractor, &st.ItemCount, &st.Error); err != nil { + return nil, err + } + out = append(out, st) + } + return out, rows.Err() +} + +func (p *pgStore) SetEdgeReview(ctx context.Context, edgeID int64, review string) error { + tag, err := p.pool.Exec(ctx, `UPDATE project_trace_edges SET review_state = $2 WHERE id = $1`, edgeID, review) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return fmt.Errorf("traceimpact: edge not found") + } + return nil +} + +func (p *pgStore) AddDecision(ctx context.Context, d Decision) (Decision, error) { + if d.ProjectID == 0 { + _ = p.pool.QueryRow(ctx, `SELECT project_id FROM project_trace_edges WHERE id = $1`, d.EdgeID).Scan(&d.ProjectID) + } + err := p.pool.QueryRow(ctx, ` + INSERT INTO project_trace_edge_decisions (project_id, edge_id, decision, reason, decided_by) + VALUES ($1,$2,$3,$4,$5) RETURNING id + `, d.ProjectID, d.EdgeID, d.Decision, d.Reason, d.DecidedBy).Scan(&d.ID) + return d, err +} + +func (p *pgStore) PutReopen(ctx context.Context, req ReopenRequest) (ReopenRequest, error) { + if req.ID == 0 { + err := p.pool.QueryRow(ctx, ` + INSERT INTO stage_reopen_requests (project_id, instance_id, stage_key, doc_task_keys, reason, requested_by, status) + VALUES ($1,$2,$3,$4,$5,$6,$7) RETURNING id + `, req.ProjectID, req.InstanceID, req.StageKey, req.DocTasks, req.Reason, req.RequestedBy, req.Status).Scan(&req.ID) + return req, err + } + _, err := p.pool.Exec(ctx, `UPDATE stage_reopen_requests SET status = $2 WHERE id = $1`, req.ID, req.Status) + return req, err +} + +func (p *pgStore) GetReopen(ctx context.Context, id int64) (ReopenRequest, error) { + var req ReopenRequest + err := p.pool.QueryRow(ctx, ` + SELECT id, project_id, instance_id, stage_key, doc_task_keys, reason, requested_by, status + FROM stage_reopen_requests WHERE id = $1 + `, id).Scan(&req.ID, &req.ProjectID, &req.InstanceID, &req.StageKey, &req.DocTasks, &req.Reason, &req.RequestedBy, &req.Status) + if err == pgx.ErrNoRows { + return ReopenRequest{}, fmt.Errorf("traceimpact: reopen request not found") + } + return req, err +} + +func (p *pgStore) PutMark(ctx context.Context, mark RecheckMark) (RecheckMark, error) { + if mark.Status == "" { + mark.Status = MarkOpen + } + err := p.pool.QueryRow(ctx, ` + INSERT INTO stage_recheck_marks (project_id, stage_key, source_doc_id, status, reason) + VALUES ($1,$2,$3,$4,$5) + ON CONFLICT (project_id, stage_key, source_doc_id) WHERE status = 'open' + DO UPDATE SET reason = stage_recheck_marks.reason + RETURNING id, project_id, stage_key, source_doc_id, status, reason, COALESCE(acked_by, 0) + `, mark.ProjectID, mark.StageKey, mark.SourceDoc, mark.Status, mark.Reason).Scan( + &mark.ID, &mark.ProjectID, &mark.StageKey, &mark.SourceDoc, &mark.Status, &mark.Reason, &mark.AckedBy, + ) + return mark, err +} + +func (p *pgStore) ListOpenMarks(ctx context.Context, projectID int64, stageKey string) ([]RecheckMark, error) { + rows, err := p.pool.Query(ctx, ` + SELECT id, project_id, stage_key, source_doc_id, status, reason, COALESCE(acked_by, 0) + FROM stage_recheck_marks + WHERE project_id = $1 AND status = 'open' AND ($2 = '' OR stage_key = $2) + ORDER BY id + `, projectID, stageKey) + if err != nil { + return nil, err + } + defer rows.Close() + var out []RecheckMark + for rows.Next() { + var m RecheckMark + if err := rows.Scan(&m.ID, &m.ProjectID, &m.StageKey, &m.SourceDoc, &m.Status, &m.Reason, &m.AckedBy); err != nil { + return nil, err + } + out = append(out, m) + } + return out, rows.Err() +} + +func (p *pgStore) AckMark(ctx context.Context, id, userID int64) error { + tag, err := p.pool.Exec(ctx, `UPDATE stage_recheck_marks SET acked_by = $2 WHERE id = $1 AND status = 'open'`, id, userID) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return fmt.Errorf("traceimpact: mark not found") + } + return nil +} + +func (p *pgStore) SetMarkStatus(ctx context.Context, id int64, status, reason string) error { + tag, err := p.pool.Exec(ctx, `UPDATE stage_recheck_marks SET status = $2, reason = CASE WHEN $3 <> '' THEN $3 ELSE reason END WHERE id = $1`, id, status, reason) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return fmt.Errorf("traceimpact: mark not found") + } + return nil +} + +func (p *pgStore) GetEdge(ctx context.Context, edgeID int64) (Edge, error) { + var e Edge + err := p.pool.QueryRow(ctx, ` + SELECT id, project_id, from_item_id, to_item_id, relation_type, provenance, review_state, evidence_doc_id + FROM project_trace_edges WHERE id = $1 + `, edgeID).Scan(&e.ID, &e.ProjectID, &e.FromItemID, &e.ToItemID, &e.RelationType, &e.Provenance, &e.ReviewState, &e.EvidenceDoc) + if err == pgx.ErrNoRows { + return Edge{}, fmt.Errorf("traceimpact: edge not found") + } + return e, err +} + +func (p *pgStore) AppendRevision(ctx context.Context, rev ItemRevision) error { + if rev.Extractor == "" { + rev.Extractor = ExtractorVersion + } + _, err := p.pool.Exec(ctx, ` + INSERT INTO project_trace_item_revisions + (item_id, revision_no, source_doc_id, source_snapshot_id, content_hash, statement, extractor_version, tombstone) + SELECT $1, + COALESCE((SELECT MAX(revision_no) + 1 FROM project_trace_item_revisions WHERE item_id = $1), 1), + $2, $3, $4, $5, $6, $7 + WHERE NOT EXISTS ( + SELECT 1 FROM project_trace_item_revisions + WHERE item_id = $1 AND source_snapshot_id = $3 AND content_hash = $4 AND extractor_version = $6 + ) + `, rev.ItemID, rev.SourceDocID, rev.SnapshotID, rev.ContentHash, rev.Statement, rev.Extractor, rev.Tombstone) + return err +} + +func (p *pgStore) RevisionCount(ctx context.Context, itemID int64) (int, error) { + var n int + err := p.pool.QueryRow(ctx, `SELECT COUNT(*) FROM project_trace_item_revisions WHERE item_id = $1`, itemID).Scan(&n) + return n, err +} + +func (p *pgStore) AddProposal(ctx context.Context, projectID, userID int64, text string) (int64, error) { + var id int64 + err := p.pool.QueryRow(ctx, ` + INSERT INTO trace_revision_proposals (project_id, author_id, body) VALUES ($1,$2,$3) RETURNING id + `, projectID, userID, text).Scan(&id) + return id, err +} diff --git a/internal/traceimpact/policy.go b/internal/traceimpact/policy.go new file mode 100644 index 00000000..dfa276b9 --- /dev/null +++ b/internal/traceimpact/policy.go @@ -0,0 +1,114 @@ +package traceimpact + +import "fmt" + +// Actor is the person trying to confirm or reject a candidate edge. +type Actor struct { + UserID int64 + Creator bool + AuthorOf map[string]bool +} + +// CanDecide reports whether actor may confirm or reject a model candidate +// on the downstream document. There is no separate trace_reviewer role. +func CanDecide(actor Actor, downstreamDocID string, downstreamAuthorID int64) error { + if downstreamDocID == "" { + return fmt.Errorf("traceimpact: missing downstream document") + } + if downstreamAuthorID > 0 { + if actor.UserID == downstreamAuthorID { + return nil + } + return fmt.Errorf("traceimpact: only the downstream author may decide this candidate") + } + if actor.Creator { + return nil + } + return fmt.Errorf("traceimpact: no downstream author; only the project creator may decide") +} + +// DutySeat is one document's author, reviewer, and approver. +type DutySeat struct { + TaskKey string + AuthorID int64 + ReviewerID int64 + ApproverID int64 +} + +// MayRequestReopen allows the project creator or a duty holder on a selected document. +func MayRequestReopen(userID int64, creator bool, selected []string, seats []DutySeat) bool { + if creator { + return true + } + wanted := map[string]bool{} + for _, key := range selected { + wanted[key] = true + } + for _, seat := range seats { + if !wanted[seat.TaskKey] { + continue + } + if userID > 0 && (userID == seat.AuthorID || userID == seat.ReviewerID || userID == seat.ApproverID) { + return true + } + } + return false +} + +// Suggest returns the three allowed next steps. They do not change state by themselves. +func Suggest(coverage string, hasImpact bool) []Suggestion { + switch coverage { + case ImpactFailed, ImpactIndexPending, ImpactPartial: + return []Suggestion{{ + Action: "wait_index", + Reason: "索引未完成或分析失败，不能判断为无影响", + }} + } + out := []Suggestion{{ + Action: ClearNoChange, + Reason: "确认这次修改不需要改其他文档", + }} + if hasImpact || coverage == ImpactPossible { + out = append(out, + Suggestion{Action: "l1_revise", Reason: "只修订受影响文档，阶段保持完成"}, + Suggestion{Action: "l2_reopen", Reason: "需要重跑编制或补充文档时再打开阶段"}, + ) + } + return out +} + +// SummaryForCoverage is the sentence shown when there is no confirmed hit. +func SummaryForCoverage(coverage string) string { + switch coverage { + case ImpactNoKnown: + return "索引已覆盖，未发现已知关联" + case ImpactIndexPending: + return "分析尚未完成，不能判断为无影响" + case ImpactPartial: + return "结果不完整，不能判断为无影响" + case ImpactFailed: + return "分析失败，不能判断为无影响" + case ImpactPossible: + return "存在未确认的可能影响" + case ImpactDependencyUnchanged: + return "已有关联，相关片段未变化" + default: + return "存在已确认影响" + } +} + +// RedactItems drops statements from documents the viewer cannot see. +// Hidden items are counted and not returned, so titles cannot leak. +func RedactItems(items []Item, visible func(docID string) bool) (out []Item, hidden int) { + for _, item := range items { + if item.SourceDocID != "" && (visible == nil || !visible(item.SourceDocID)) { + hidden++ + continue + } + out = append(out, item) + } + if out == nil { + out = []Item{} + } + return out, hidden +} diff --git a/internal/traceimpact/service.go b/internal/traceimpact/service.go new file mode 100644 index 00000000..19cc9c81 --- /dev/null +++ b/internal/traceimpact/service.go @@ -0,0 +1,380 @@ +package traceimpact + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "strings" +) + +// StageReopener applies an approved reopen to the workflow instance. +// It must not pause the instance or rewrite downstream stages. +type StageReopener interface { + ReopenCompletedStage(ctx context.Context, instanceID int64, stageKey string, docTaskKeys []string, operatorID int64) error +} + +// DocUnlocker moves selected approved documents back to drafting. +type DocUnlocker interface { + ReopenApprovedDocs(ctx context.Context, projectID int64, docTaskKeys []string, actorID int64, reason string) error +} + +// Service is the project trace index and the stage-backtrack ledger. +type Service struct { + store Store + reopen StageReopener + unlock DocUnlocker +} + +func NewService(store Store) *Service { + if store == nil { + store = newMemStore() + } + return &Service{store: store} +} + +func NewMemoryService() *Service { + return NewService(newMemStore()) +} + +func (s *Service) SetStageReopener(r StageReopener) { s.reopen = r } +func (s *Service) SetDocUnlocker(u DocUnlocker) { s.unlock = u } + +// IndexDocument projects explicit keys into items and same-line pairs into confirmed edges. +// Untyped IDs become candidate items only. +func (s *Service) IndexDocument(ctx context.Context, projectID int64, doc DocRef) (IndexState, error) { + if projectID <= 0 || doc.DocID == "" { + return IndexState{}, fmt.Errorf("traceimpact: project and doc are required") + } + keys := ExtractKeys(doc.Markdown) + itemByKey := map[string]Item{} + for _, k := range keys { + review := ReviewConfirmed + if !k.Explicit { + review = ReviewCandidate + } + item, err := s.store.UpsertItem(ctx, Item{ + ProjectID: projectID, BusinessKey: k.Key, Kind: k.Kind, + ReviewState: review, Statement: strings.TrimSpace(k.Line), SourceDocID: doc.DocID, + }) + if err != nil { + st := IndexState{ProjectID: projectID, DocID: doc.DocID, SnapshotID: doc.SnapshotID, Status: ImpactFailed, Error: err.Error()} + _ = s.store.PutIndex(ctx, st) + return st, err + } + sum := sha256.Sum256([]byte(k.Key + "\n" + strings.TrimSpace(k.Line))) + if err := s.store.AppendRevision(ctx, ItemRevision{ + ItemID: item.ID, SourceDocID: doc.DocID, SnapshotID: doc.SnapshotID, + ContentHash: hex.EncodeToString(sum[:]), Statement: strings.TrimSpace(k.Line), + Extractor: ExtractorVersion, + }); err != nil { + return IndexState{}, err + } + itemByKey[k.Key] = item + } + for _, pair := range ExplicitPairsOnLine(doc.Markdown) { + from, okA := itemByKey[pair[0]] + to, okB := itemByKey[pair[1]] + if !okA || !okB { + continue + } + if _, err := s.store.UpsertEdge(ctx, Edge{ + ProjectID: projectID, FromItemID: from.ID, ToItemID: to.ID, + RelationType: RelationRefines, Provenance: ProvenanceExplicit, + ReviewState: ReviewConfirmed, EvidenceDoc: doc.DocID, + FromKey: from.BusinessKey, ToKey: to.BusinessKey, + }); err != nil { + return IndexState{}, err + } + } + st := IndexState{ + ProjectID: projectID, DocID: doc.DocID, SnapshotID: doc.SnapshotID, + Status: "complete", Extractor: ExtractorVersion, ItemCount: len(keys), + } + if err := s.store.PutIndex(ctx, st); err != nil { + return IndexState{}, err + } + return st, nil +} + +// CoverageForDoc is what a change report should show before claiming no impact. +func (s *Service) CoverageForDoc(ctx context.Context, projectID int64, docID string) string { + st, ok, err := s.store.IndexStatus(ctx, projectID, docID) + if err != nil || !ok || st.Status == "" { + return ImpactIndexPending + } + switch st.Status { + case "complete": + return ImpactNoKnown + case "failed": + return ImpactFailed + case "partial": + return ImpactPartial + default: + return ImpactIndexPending + } +} + +// Analyze walks confirmed edges from keys mentioned in the changed markdown. +// Missing index coverage never becomes "no impact". +func (s *Service) Analyze(ctx context.Context, projectID int64, source DocRef, others []DocRef) (*Report, error) { + rep := &Report{ProjectID: projectID, SourceDocID: source.DocID, Impacts: []Impact{}} + if projectID <= 0 || source.DocID == "" { + rep.CoverageStatus = ImpactFailed + rep.Summary = SummaryForCoverage(rep.CoverageStatus) + rep.Suggestions = Suggest(rep.CoverageStatus, false) + return rep, nil + } + srcIndex, ok, err := s.store.IndexStatus(ctx, projectID, source.DocID) + if err != nil { + rep.CoverageStatus = ImpactFailed + rep.Summary = SummaryForCoverage(rep.CoverageStatus) + rep.Suggestions = Suggest(rep.CoverageStatus, false) + return rep, nil + } + if !ok || srcIndex.Status != "complete" { + rep.CoverageStatus = ImpactIndexPending + if ok && srcIndex.Status == "failed" { + rep.CoverageStatus = ImpactFailed + } + rep.Summary = SummaryForCoverage(rep.CoverageStatus) + rep.Suggestions = Suggest(rep.CoverageStatus, false) + return rep, nil + } + + items, err := s.store.ListItems(ctx, projectID) + if err != nil { + rep.CoverageStatus = ImpactFailed + rep.Summary = SummaryForCoverage(rep.CoverageStatus) + rep.Suggestions = Suggest(rep.CoverageStatus, false) + return rep, nil + } + byID := map[int64]Item{} + byKey := map[string]Item{} + for _, it := range items { + byID[it.ID] = it + byKey[it.BusinessKey] = it + } + var start []int64 + for _, k := range ExtractKeys(source.Markdown) { + if it, ok := byKey[k.Key]; ok && it.ReviewState == ReviewConfirmed { + start = append(start, it.ID) + } + } + edges, err := s.store.ListEdges(ctx, projectID) + if err != nil { + rep.CoverageStatus = ImpactFailed + rep.Summary = SummaryForCoverage(rep.CoverageStatus) + rep.Suggestions = Suggest(rep.CoverageStatus, false) + return rep, nil + } + hits, truncated := WalkReverse(edges, start, MaxWalkDepth, MaxWalkNodes) + docByKey := map[string]DocRef{source.DocID: source} + for _, d := range others { + docByKey[d.DocID] = d + } + for _, hit := range hits { + it := byID[hit.ItemID] + if !it.Tombstone && it.SourceDocID != "" && it.SourceDocID != source.DocID { + target := docByKey[it.SourceDocID] + if target.DocID != "" && !target.Visible { + continue + } + rep.Impacts = append(rep.Impacts, Impact{ + DocID: it.SourceDocID, ItemKey: it.BusinessKey, StageKey: target.StageKey, + ImpactType: ImpactConfirmed, Depth: hit.Depth, + Reason: fmt.Sprintf("%s 经已确认关系受本次修改影响", it.BusinessKey), + }) + if target.StageStatus == "completed" && target.StageKey != "" { + _, _ = s.store.PutMark(ctx, RecheckMark{ + ProjectID: projectID, StageKey: target.StageKey, SourceDoc: source.DocID, + Status: MarkOpen, Reason: "上游或同项目条目已变化", + }) + } + } + } + for _, e := range edges { + if e.ReviewState != ReviewCandidate { + continue + } + for _, id := range start { + if e.ToItemID == id || e.FromItemID == id { + rep.Impacts = append(rep.Impacts, Impact{ + ItemKey: byID[e.FromItemID].BusinessKey, ImpactType: ImpactPossible, Depth: 1, + Reason: "候选关系尚未确认", + }) + break + } + } + } + switch { + case truncated: + rep.CoverageStatus = ImpactPartial + case len(rep.Impacts) == 0: + rep.CoverageStatus = ImpactNoKnown + default: + rep.CoverageStatus = ImpactConfirmed + for _, imp := range rep.Impacts { + if imp.ImpactType == ImpactPossible && rep.CoverageStatus != ImpactConfirmed { + rep.CoverageStatus = ImpactPossible + } + if imp.ImpactType == ImpactConfirmed { + rep.CoverageStatus = ImpactConfirmed + break + } + } + onlyPossible := true + for _, imp := range rep.Impacts { + if imp.ImpactType == ImpactConfirmed { + onlyPossible = false + } + } + if onlyPossible { + rep.CoverageStatus = ImpactPossible + } + } + rep.Summary = SummaryForCoverage(rep.CoverageStatus) + if rep.CoverageStatus == ImpactConfirmed { + rep.Summary = fmt.Sprintf("发现 %d 条影响", len(rep.Impacts)) + } + rep.Suggestions = Suggest(rep.CoverageStatus, len(rep.Impacts) > 0) + return rep, nil +} + +// DecideCandidate confirms or rejects a model/candidate edge. +func (s *Service) DecideCandidate(ctx context.Context, edgeID, actorID int64, creator bool, downstreamDoc string, downstreamAuthor int64, decision, reason string) error { + if err := CanDecide(Actor{UserID: actorID, Creator: creator}, downstreamDoc, downstreamAuthor); err != nil { + return err + } + if decision != DecisionConfirm && decision != DecisionReject { + return fmt.Errorf("traceimpact: decision must be confirm or reject") + } + edge, err := s.store.GetEdge(ctx, edgeID) + if err != nil { + return err + } + review := ReviewConfirmed + if decision == DecisionReject { + review = ReviewRejected + } + if err := s.store.SetEdgeReview(ctx, edgeID, review); err != nil { + return err + } + _, err = s.store.AddDecision(ctx, Decision{ + ProjectID: edge.ProjectID, EdgeID: edgeID, Decision: decision, Reason: reason, DecidedBy: actorID, + }) + return err +} + +// NeedsRecheckAck is true when the stage still has an open mark. +// Acknowledgement does not clear the mark. +func (s *Service) NeedsRecheckAck(ctx context.Context, projectID int64, stageKey string) (bool, error) { + marks, err := s.store.ListOpenMarks(ctx, projectID, stageKey) + if err != nil { + return false, err + } + return len(marks) > 0, nil +} + +// AcknowledgeRecheck records that the creator saw the marks. Marks stay open. +func (s *Service) AcknowledgeRecheck(ctx context.Context, projectID int64, stageKey string, userID int64) error { + marks, err := s.store.ListOpenMarks(ctx, projectID, stageKey) + if err != nil { + return err + } + for _, m := range marks { + if err := s.store.AckMark(ctx, m.ID, userID); err != nil { + return err + } + } + return nil +} + +// ClearRecheck closes a mark only for an explicit human reason. +func (s *Service) ClearRecheck(ctx context.Context, markID int64, reason string) error { + switch reason { + case ClearNoChange, ClearRevision: + default: + return fmt.Errorf("traceimpact: recheck marks close only by no_change or revision_started") + } + return s.store.SetMarkStatus(ctx, markID, MarkCleared, reason) +} + +// ReopenMark opens a previously cleared mark again, for example after a rejected proposal. +func (s *Service) ReopenMark(ctx context.Context, markID int64, reason string) error { + return s.store.SetMarkStatus(ctx, markID, MarkOpen, reason) +} + +// RequestReopen stores a request. Creators are approved immediately and applied. +func (s *Service) RequestReopen(ctx context.Context, req ReopenRequest, creator bool) (ReopenRequest, error) { + if req.StageKey == "" || len(req.DocTasks) == 0 || strings.TrimSpace(req.Reason) == "" { + return ReopenRequest{}, fmt.Errorf("traceimpact: stage, documents, and reason are required") + } + req.Status = ReopenPending + stored, err := s.store.PutReopen(ctx, req) + if err != nil { + return ReopenRequest{}, err + } + if creator { + return s.ApproveReopen(ctx, stored.ID, req.RequestedBy) + } + return stored, nil +} + +// ApproveReopen lets the creator apply a pending request. +func (s *Service) ApproveReopen(ctx context.Context, requestID, creatorID int64) (ReopenRequest, error) { + req, err := s.store.GetReopen(ctx, requestID) + if err != nil { + return ReopenRequest{}, err + } + if req.Status == ReopenApproved { + return req, nil + } + if s.reopen == nil { + return ReopenRequest{}, fmt.Errorf("traceimpact: stage reopen is not configured") + } + if err := s.reopen.ReopenCompletedStage(ctx, req.InstanceID, req.StageKey, req.DocTasks, creatorID); err != nil { + return ReopenRequest{}, err + } + if s.unlock != nil { + if err := s.unlock.ReopenApprovedDocs(ctx, req.ProjectID, req.DocTasks, creatorID, req.Reason); err != nil { + return ReopenRequest{}, err + } + } + req.Status = ReopenApproved + return s.store.PutReopen(ctx, req) +} + +func (s *Service) ListItems(ctx context.Context, projectID int64) ([]Item, error) { + return s.store.ListItems(ctx, projectID) +} + +func (s *Service) ListIndex(ctx context.Context, projectID int64) ([]IndexState, error) { + return s.store.ListIndex(ctx, projectID) +} + +func (s *Service) ListOpenMarks(ctx context.Context, projectID int64, stageKey string) ([]RecheckMark, error) { + return s.store.ListOpenMarks(ctx, projectID, stageKey) +} + +func (s *Service) GetReopen(ctx context.Context, id int64) (ReopenRequest, error) { + return s.store.GetReopen(ctx, id) +} + +// ErrNeedsConfirm is returned when a revision proposal is requested before confirmation. +var ErrNeedsConfirm = fmt.Errorf("traceimpact: confirmation required") + +// ProposeRevision records a proposal for a human author. It never edits controlled content. +func (s *Service) ProposeRevision(ctx context.Context, projectID, userID int64, text string, confirmed bool) (string, error) { + if !confirmed { + return "", ErrNeedsConfirm + } + if projectID <= 0 || strings.TrimSpace(text) == "" { + return "", fmt.Errorf("traceimpact: project and proposal text are required") + } + id, err := s.store.AddProposal(ctx, projectID, userID, strings.TrimSpace(text)) + if err != nil { + return "", err + } + return fmt.Sprintf("proposal-%d-by-%d", id, userID), nil +} diff --git a/internal/traceimpact/service_test.go b/internal/traceimpact/service_test.go new file mode 100644 index 00000000..1847a4a3 --- /dev/null +++ b/internal/traceimpact/service_test.go @@ -0,0 +1,230 @@ +package traceimpact + +import ( + "context" + "strings" + "testing" +) + +func TestExtract_UntypedIsCandidateOnly(t *testing.T) { + keys := ExtractKeys("UR-001 对应 RS-002\n另见 AB-9") + explicit, untyped := 0, 0 + for _, k := range keys { + if k.Explicit { + explicit++ + } else { + untyped++ + } + } + if explicit != 2 || untyped != 1 { + t.Fatalf("explicit=%d untyped=%d keys=%v", explicit, untyped, keys) + } +} + +func TestWalkReverse_DepthAndCycle(t *testing.T) { + edges := []Edge{ + {ID: 1, FromItemID: 2, ToItemID: 1, RelationType: RelationRefines, ReviewState: ReviewConfirmed}, + {ID: 2, FromItemID: 3, ToItemID: 2, RelationType: RelationDependsOn, ReviewState: ReviewConfirmed}, + {ID: 3, FromItemID: 4, ToItemID: 3, RelationType: RelationDependsOn, ReviewState: ReviewConfirmed}, + {ID: 4, FromItemID: 1, ToItemID: 4, RelationType: RelationDependsOn, ReviewState: ReviewConfirmed}, + {ID: 5, FromItemID: 9, ToItemID: 1, RelationType: RelationDependsOn, ReviewState: ReviewCandidate}, + } + hits, truncated := WalkReverse(edges, []int64{1}, 3, 50) + if truncated { + t.Fatal("did not expect truncation") + } + got := map[int64]int{} + for _, h := range hits { + got[h.ItemID] = h.Depth + } + if got[2] != 1 || got[3] != 2 || got[4] != 3 { + t.Fatalf("depths = %#v", got) + } + if _, ok := got[9]; ok { + t.Fatal("candidate edge must not be followed") + } + if _, ok := got[1]; ok { + t.Fatal("cycle must not revisit the start as a hit") + } +} + +func TestAnalyze_PendingIsNotNoImpact(t *testing.T) { + svc := NewMemoryService() + rep, err := svc.Analyze(t.Context(), 7, DocRef{DocID: "doc-a", Markdown: "UR-001"}, nil) + if err != nil { + t.Fatal(err) + } + if rep.CoverageStatus != ImpactIndexPending { + t.Fatalf("status %s", rep.CoverageStatus) + } + if strings.Contains(rep.Summary, "未发现") { + t.Fatalf("pending must not look like no impact: %s", rep.Summary) + } +} + +func TestAnalyze_ConfirmedDownstreamAndRecheck(t *testing.T) { + svc := NewMemoryService() + ctx := t.Context() + if _, err := svc.IndexDocument(ctx, 7, DocRef{DocID: "up", Markdown: "UR-001 用户需要记录"}); err != nil { + t.Fatal(err) + } + if _, err := svc.IndexDocument(ctx, 7, DocRef{DocID: "down", Markdown: "RS-010 细化 UR-001"}); err != nil { + t.Fatal(err) + } + rep, err := svc.Analyze(ctx, 7, DocRef{DocID: "up", Markdown: "UR-001 改为必须记录"}, []DocRef{{ + DocID: "down", StageKey: "V0.5", StageStatus: "completed", Visible: true, + }}) + if err != nil { + t.Fatal(err) + } + if rep.CoverageStatus != ImpactConfirmed { + t.Fatalf("status %s summary %s impacts %#v", rep.CoverageStatus, rep.Summary, rep.Impacts) + } + need, err := svc.NeedsRecheckAck(ctx, 7, "V0.5") + if err != nil || !need { + t.Fatalf("need ack %v %v", need, err) + } + if err := svc.AcknowledgeRecheck(ctx, 7, "V0.5", 3); err != nil { + t.Fatal(err) + } + need, err = svc.NeedsRecheckAck(ctx, 7, "V0.5") + if err != nil || !need { + t.Fatal("ack must not clear the mark") + } + marks, err := svc.store.ListOpenMarks(ctx, 7, "V0.5") + if err != nil || len(marks) != 1 { + t.Fatal(err) + } + if err := svc.ClearRecheck(ctx, marks[0].ID, "auto"); err == nil { + t.Fatal("auto clear must fail") + } + if err := svc.ClearRecheck(ctx, marks[0].ID, ClearNoChange); err != nil { + t.Fatal(err) + } + need, _ = svc.NeedsRecheckAck(ctx, 7, "V0.5") + if need { + t.Fatal("explicit no_change should clear") + } +} + +func TestDecide_DownstreamAuthorOnly(t *testing.T) { + if err := CanDecide(Actor{UserID: 2, Creator: true}, "down", 9); err == nil { + t.Fatal("creator must not override an existing author") + } + if err := CanDecide(Actor{UserID: 9}, "down", 9); err != nil { + t.Fatal(err) + } + if err := CanDecide(Actor{UserID: 1, Creator: true}, "down", 0); err != nil { + t.Fatal(err) + } +} + +type fakeReopen struct{ called bool } + +func (f *fakeReopen) ReopenCompletedStage(_ context.Context, _ int64, stageKey string, docTaskKeys []string, _ int64) error { + f.called = true + if stageKey != "V0.1" || len(docTaskKeys) != 1 { + return fmtError("bad reopen") + } + return nil +} + +type fmtError string + +func (e fmtError) Error() string { return string(e) } + +func TestReopen_CreatorAppliesWithoutPausingHook(t *testing.T) { + svc := NewMemoryService() + hook := &fakeReopen{} + svc.SetStageReopener(hook) + req, err := svc.RequestReopen(t.Context(), ReopenRequest{ + ProjectID: 1, InstanceID: 2, StageKey: "V0.1", DocTasks: []string{"bd_intake"}, + Reason: "新材料推翻了摸底", RequestedBy: 4, + }, true) + if err != nil { + t.Fatal(err) + } + if req.Status != ReopenApproved || !hook.called { + t.Fatalf("status %s called %v", req.Status, hook.called) + } +} + +func TestReopen_MemberStaysPending(t *testing.T) { + svc := NewMemoryService() + req, err := svc.RequestReopen(t.Context(), ReopenRequest{ + ProjectID: 1, InstanceID: 2, StageKey: "V0.1", DocTasks: []string{"bd_intake"}, + Reason: "需要重编", RequestedBy: 8, + }, false) + if err != nil { + t.Fatal(err) + } + if err != nil || req.Status != ReopenPending { + t.Fatalf("status %s err %v", req.Status, err) + } +} + +func TestIndex_SameSnapshotDoesNotDuplicateRevision(t *testing.T) { + svc := NewMemoryService() + doc := DocRef{DocID: "doc_proj1_intake", Markdown: "UR-001 用户需求", SnapshotID: 7, Visible: true} + if _, err := svc.IndexDocument(t.Context(), 1, doc); err != nil { + t.Fatal(err) + } + if _, err := svc.IndexDocument(t.Context(), 1, doc); err != nil { + t.Fatal(err) + } + items, err := svc.ListItems(t.Context(), 1) + if err != nil || len(items) != 1 { + t.Fatalf("items %+v err %v", items, err) + } + n, err := svc.store.RevisionCount(t.Context(), items[0].ID) + if err != nil || n != 1 { + t.Fatalf("revisions %d err %v", n, err) + } + again, err := svc.IndexDocument(t.Context(), 2, DocRef{DocID: "doc_proj2_intake", Markdown: "UR-001 另一项目", Visible: true}) + if err != nil { + t.Fatal(err) + } + _ = again + other, err := svc.ListItems(t.Context(), 1) + if err != nil || len(other) != 1 || other[0].ProjectID != 1 { + t.Fatalf("cross project leaked %+v", other) + } +} + +func TestRedactItems_HidesInvisibleStatements(t *testing.T) { + items := []Item{ + {BusinessKey: "UR-001", Statement: "可见标题", SourceDocID: "doc_proj1_intake"}, + {BusinessKey: "REQ-9", Statement: "秘密标题", SourceDocID: "doc_proj1_secret"}, + } + out, hidden := RedactItems(items, func(docID string) bool { return docID == "doc_proj1_intake" }) + if hidden != 1 || len(out) != 1 || out[0].Statement != "可见标题" { + t.Fatalf("out %+v hidden %d", out, hidden) + } + blob := strings.Join([]string{out[0].BusinessKey, out[0].Statement}, " ") + if strings.Contains(blob, "秘密") { + t.Fatal(blob) + } +} + +func TestMayRequestReopen_DutyHolderOnly(t *testing.T) { + seats := []DutySeat{{TaskKey: "intake", AuthorID: 8, ReviewerID: 2, ApproverID: 3}} + if !MayRequestReopen(8, false, []string{"intake"}, seats) { + t.Fatal("author") + } + if !MayRequestReopen(1, true, []string{"intake"}, nil) { + t.Fatal("creator") + } + if MayRequestReopen(99, false, []string{"intake"}, seats) { + t.Fatal("stranger") + } +} + +func TestPropose_UnconfirmedDoesNotStore(t *testing.T) { + svc := NewMemoryService() + if _, err := svc.ProposeRevision(t.Context(), 1, 4, "改写 UR-001", false); err != ErrNeedsConfirm { + t.Fatalf("err %v", err) + } + if _, err := svc.ProposeRevision(t.Context(), 1, 4, "改写 UR-001", true); err != nil { + t.Fatal(err) + } +} diff --git a/internal/traceimpact/store.go b/internal/traceimpact/store.go new file mode 100644 index 00000000..183d3150 --- /dev/null +++ b/internal/traceimpact/store.go @@ -0,0 +1,317 @@ +package traceimpact + +import ( + "context" + "fmt" + "sync" +) + +// Store persists trace rows. Implementations must not delete decision history. +type Store interface { + UpsertItem(ctx context.Context, item Item) (Item, error) + ListItems(ctx context.Context, projectID int64) ([]Item, error) + UpsertEdge(ctx context.Context, edge Edge) (Edge, error) + ListEdges(ctx context.Context, projectID int64) ([]Edge, error) + PutIndex(ctx context.Context, st IndexState) error + IndexStatus(ctx context.Context, projectID int64, docID string) (IndexState, bool, error) + ListIndex(ctx context.Context, projectID int64) ([]IndexState, error) + SetEdgeReview(ctx context.Context, edgeID int64, review string) error + AddDecision(ctx context.Context, d Decision) (Decision, error) + PutReopen(ctx context.Context, req ReopenRequest) (ReopenRequest, error) + GetReopen(ctx context.Context, id int64) (ReopenRequest, error) + PutMark(ctx context.Context, m RecheckMark) (RecheckMark, error) + ListOpenMarks(ctx context.Context, projectID int64, stageKey string) ([]RecheckMark, error) + AckMark(ctx context.Context, id, userID int64) error + SetMarkStatus(ctx context.Context, id int64, status, reason string) error + GetEdge(ctx context.Context, edgeID int64) (Edge, error) + AppendRevision(ctx context.Context, rev ItemRevision) error + RevisionCount(ctx context.Context, itemID int64) (int, error) + AddProposal(ctx context.Context, projectID, userID int64, text string) (int64, error) +} + +type memStore struct { + mu sync.Mutex + items map[int64]Item + byKey map[string]int64 + edges map[int64]Edge + byEdge map[string]int64 + index map[string]IndexState + decisions []Decision + reopens map[int64]ReopenRequest + marks map[int64]RecheckMark + revs map[int64][]ItemRevision + props []proposalRow + nextItem int64 + nextEdge int64 + nextDec int64 + nextReq int64 + nextMark int64 + nextProp int64 +} + +func newMemStore() *memStore { + return &memStore{ + items: map[int64]Item{}, + byKey: map[string]int64{}, + edges: map[int64]Edge{}, + byEdge: map[string]int64{}, + index: map[string]IndexState{}, + reopens: map[int64]ReopenRequest{}, + marks: map[int64]RecheckMark{}, + } +} + +func itemKey(projectID int64, kind, business string) string { + return fmt.Sprintf("%d|%s|%s", projectID, kind, business) +} + +type proposalRow struct { + id int64 + projectID int64 + userID int64 + text string +} + +func (m *memStore) UpsertItem(_ context.Context, item Item) (Item, error) { + m.mu.Lock() + defer m.mu.Unlock() + k := itemKey(item.ProjectID, item.Kind, item.BusinessKey) + if id, ok := m.byKey[k]; ok { + cur := m.items[id] + if item.Statement != "" { + cur.Statement = item.Statement + } + if item.SourceDocID != "" { + cur.SourceDocID = item.SourceDocID + } + cur.ReviewState = item.ReviewState + cur.Tombstone = item.Tombstone + m.items[id] = cur + return cur, nil + } + m.nextItem++ + item.ID = m.nextItem + if item.Lifecycle == "" { + item.Lifecycle = "active" + } + m.items[item.ID] = item + m.byKey[k] = item.ID + return item, nil +} + +func (m *memStore) ListItems(_ context.Context, projectID int64) ([]Item, error) { + m.mu.Lock() + defer m.mu.Unlock() + var out []Item + for _, it := range m.items { + if it.ProjectID == projectID { + out = append(out, it) + } + } + return out, nil +} + +func (m *memStore) UpsertEdge(_ context.Context, edge Edge) (Edge, error) { + m.mu.Lock() + defer m.mu.Unlock() + k := fmt.Sprintf("%d|%d|%d|%s|%s", edge.ProjectID, edge.FromItemID, edge.ToItemID, edge.RelationType, edge.Provenance) + if id, ok := m.byEdge[k]; ok { + cur := m.edges[id] + if cur.ReviewState == ReviewRejected && edge.ReviewState == ReviewCandidate { + return cur, nil + } + if edge.ReviewState != "" && cur.ReviewState != ReviewConfirmed { + cur.ReviewState = edge.ReviewState + } + m.edges[id] = cur + return cur, nil + } + m.nextEdge++ + edge.ID = m.nextEdge + m.edges[edge.ID] = edge + m.byEdge[k] = edge.ID + return edge, nil +} + +func (m *memStore) ListEdges(_ context.Context, projectID int64) ([]Edge, error) { + m.mu.Lock() + defer m.mu.Unlock() + var out []Edge + for _, e := range m.edges { + if e.ProjectID == projectID { + out = append(out, e) + } + } + return out, nil +} + +func (m *memStore) PutIndex(_ context.Context, st IndexState) error { + m.mu.Lock() + defer m.mu.Unlock() + if st.Extractor == "" { + st.Extractor = ExtractorVersion + } + m.index[fmt.Sprintf("%d|%s", st.ProjectID, st.DocID)] = st + return nil +} + +func (m *memStore) IndexStatus(_ context.Context, projectID int64, docID string) (IndexState, bool, error) { + m.mu.Lock() + defer m.mu.Unlock() + st, ok := m.index[fmt.Sprintf("%d|%s", projectID, docID)] + return st, ok, nil +} + +func (m *memStore) ListIndex(_ context.Context, projectID int64) ([]IndexState, error) { + m.mu.Lock() + defer m.mu.Unlock() + var out []IndexState + for _, st := range m.index { + if st.ProjectID == projectID { + out = append(out, st) + } + } + return out, nil +} + +func (m *memStore) SetEdgeReview(_ context.Context, edgeID int64, review string) error { + m.mu.Lock() + defer m.mu.Unlock() + e, ok := m.edges[edgeID] + if !ok { + return fmt.Errorf("traceimpact: edge not found") + } + e.ReviewState = review + m.edges[edgeID] = e + return nil +} + +func (m *memStore) AddDecision(_ context.Context, d Decision) (Decision, error) { + m.mu.Lock() + defer m.mu.Unlock() + m.nextDec++ + d.ID = m.nextDec + m.decisions = append(m.decisions, d) + return d, nil +} + +func (m *memStore) PutReopen(_ context.Context, req ReopenRequest) (ReopenRequest, error) { + m.mu.Lock() + defer m.mu.Unlock() + if req.ID == 0 { + m.nextReq++ + req.ID = m.nextReq + } + m.reopens[req.ID] = req + return req, nil +} + +func (m *memStore) GetReopen(_ context.Context, id int64) (ReopenRequest, error) { + m.mu.Lock() + defer m.mu.Unlock() + req, ok := m.reopens[id] + if !ok { + return ReopenRequest{}, fmt.Errorf("traceimpact: reopen request not found") + } + return req, nil +} + +func (m *memStore) PutMark(_ context.Context, mark RecheckMark) (RecheckMark, error) { + m.mu.Lock() + defer m.mu.Unlock() + for _, existing := range m.marks { + if existing.ProjectID == mark.ProjectID && existing.StageKey == mark.StageKey && existing.SourceDoc == mark.SourceDoc && existing.Status == MarkOpen { + return existing, nil + } + } + m.nextMark++ + mark.ID = m.nextMark + if mark.Status == "" { + mark.Status = MarkOpen + } + m.marks[mark.ID] = mark + return mark, nil +} + +func (m *memStore) ListOpenMarks(_ context.Context, projectID int64, stageKey string) ([]RecheckMark, error) { + m.mu.Lock() + defer m.mu.Unlock() + var out []RecheckMark + for _, mark := range m.marks { + if mark.ProjectID != projectID || mark.Status != MarkOpen { + continue + } + if stageKey != "" && mark.StageKey != stageKey { + continue + } + out = append(out, mark) + } + return out, nil +} + +func (m *memStore) AckMark(_ context.Context, id, userID int64) error { + m.mu.Lock() + defer m.mu.Unlock() + mark, ok := m.marks[id] + if !ok { + return fmt.Errorf("traceimpact: mark not found") + } + mark.AckedBy = userID + m.marks[id] = mark + return nil +} + +func (m *memStore) SetMarkStatus(_ context.Context, id int64, status, reason string) error { + m.mu.Lock() + defer m.mu.Unlock() + mark, ok := m.marks[id] + if !ok { + return fmt.Errorf("traceimpact: mark not found") + } + mark.Status = status + if reason != "" { + mark.Reason = reason + } + m.marks[id] = mark + return nil +} + +func (m *memStore) GetEdge(_ context.Context, edgeID int64) (Edge, error) { + m.mu.Lock() + defer m.mu.Unlock() + e, ok := m.edges[edgeID] + if !ok { + return Edge{}, fmt.Errorf("traceimpact: edge not found") + } + return e, nil +} + +func (m *memStore) AppendRevision(_ context.Context, rev ItemRevision) error { + m.mu.Lock() + defer m.mu.Unlock() + if m.revs == nil { + m.revs = map[int64][]ItemRevision{} + } + for _, existing := range m.revs[rev.ItemID] { + if existing.SnapshotID == rev.SnapshotID && existing.ContentHash == rev.ContentHash && existing.Extractor == rev.Extractor { + return nil + } + } + rev.RevisionNo = len(m.revs[rev.ItemID]) + 1 + m.revs[rev.ItemID] = append(m.revs[rev.ItemID], rev) + return nil +} + +func (m *memStore) RevisionCount(_ context.Context, itemID int64) (int, error) { + m.mu.Lock() + defer m.mu.Unlock() + return len(m.revs[itemID]), nil +} + +func (m *memStore) AddProposal(_ context.Context, projectID, userID int64, text string) (int64, error) { + m.mu.Lock() + defer m.mu.Unlock() + m.nextProp++ + m.props = append(m.props, proposalRow{id: m.nextProp, projectID: projectID, userID: userID, text: text}) + return m.nextProp, nil +} diff --git a/internal/traceimpact/types.go b/internal/traceimpact/types.go new file mode 100644 index 00000000..d3aa3a63 --- /dev/null +++ b/internal/traceimpact/types.go @@ -0,0 +1,164 @@ +package traceimpact + +// Coverage and impact statuses. Unknown, failed, and empty are different. +const ( + ImpactConfirmed = "confirmed_impact" + ImpactPossible = "possible_impact" + ImpactDependencyUnchanged = "dependency_unchanged" + ImpactNoKnown = "no_known_dependencies" + ImpactIndexPending = "index_pending" + ImpactPartial = "partial" + ImpactFailed = "failed" + + KindRequirement = "requirement" + KindTest = "test" + + ReviewConfirmed = "confirmed" + ReviewCandidate = "candidate" + ReviewRejected = "rejected" + + ProvenanceExplicit = "explicit_reference" + ProvenanceRule = "rule" + ProvenanceModel = "model_candidate" + + RelationRefines = "refines" + RelationDependsOn = "depends_on" + + DecisionConfirm = "confirm" + DecisionReject = "reject" + + ReopenPending = "pending" + ReopenApproved = "approved" + ReopenRejected = "rejected" + + MarkOpen = "open" + MarkCleared = "cleared" + + ClearNoChange = "no_change" + ClearRevision = "revision_started" + + ExtractorVersion = "trace-keys-v1" + MaxWalkDepth = 3 + MaxWalkNodes = 50 +) + +// ItemRevision is an immutable projection of one item at one snapshot. +type ItemRevision struct { + ID int64 + ItemID int64 + RevisionNo int + SourceDocID string + SnapshotID int64 + ContentHash string + Statement string + Extractor string + Tombstone bool +} + +// Item is a stable project entry. +type Item struct { + ID int64 `json:"id"` + ProjectID int64 `json:"project_id"` + BusinessKey string `json:"business_key"` + Kind string `json:"kind"` + Lifecycle string `json:"lifecycle_state"` + ReviewState string `json:"review_state"` + Tombstone bool `json:"tombstone"` + Statement string `json:"statement,omitempty"` + SourceDocID string `json:"source_doc_id,omitempty"` +} + +// Edge is a directed relation. Reverse walk follows dependents of a changed item. +type Edge struct { + ID int64 `json:"id"` + ProjectID int64 `json:"project_id"` + FromItemID int64 `json:"from_item_id"` + ToItemID int64 `json:"to_item_id"` + RelationType string `json:"relation_type"` + Provenance string `json:"provenance"` + ReviewState string `json:"review_state"` + EvidenceDoc string `json:"evidence_doc_id,omitempty"` + FromKey string `json:"from_key,omitempty"` + ToKey string `json:"to_key,omitempty"` +} + +// Decision is the authority record for a candidate edge. +type Decision struct { + ID int64 `json:"id"` + ProjectID int64 `json:"project_id"` + EdgeID int64 `json:"edge_id"` + Decision string `json:"decision"` + Reason string `json:"reason"` + DecidedBy int64 `json:"decided_by"` +} + +// IndexState is the coverage watermark for one document snapshot. +type IndexState struct { + ProjectID int64 `json:"project_id"` + DocID string `json:"doc_id"` + SnapshotID int64 `json:"source_snapshot_id"` + Status string `json:"status"` + Extractor string `json:"extractor_version"` + ItemCount int `json:"item_count"` + Error string `json:"error,omitempty"` +} + +// DocRef binds a document to the stage the caller is asking about. +type DocRef struct { + DocID string `json:"doc_id"` + StageKey string `json:"stage_key,omitempty"` + StageStatus string `json:"stage_status,omitempty"` + AuthorID int64 `json:"author_id,omitempty"` + Visible bool `json:"visible"` + Markdown string `json:"markdown,omitempty"` + SnapshotID int64 `json:"snapshot_id,omitempty"` +} + +// Suggestion is a clickable next step, not a free-text instruction. +type Suggestion struct { + Action string `json:"action"` + Reason string `json:"reason"` +} + +// Impact is one affected document or item. +type Impact struct { + DocID string `json:"doc_id,omitempty"` + ItemKey string `json:"item_key,omitempty"` + StageKey string `json:"stage_key,omitempty"` + ImpactType string `json:"impact_type"` + Depth int `json:"depth"` + Reason string `json:"reason"` +} + +// Report is the deterministic impact result. +type Report struct { + ProjectID int64 `json:"project_id"` + SourceDocID string `json:"source_doc_id"` + CoverageStatus string `json:"coverage_status"` + Impacts []Impact `json:"impacts"` + Suggestions []Suggestion `json:"suggestions"` + Summary string `json:"summary"` +} + +// ReopenRequest asks to unlock selected documents on a completed stage. +type ReopenRequest struct { + ID int64 `json:"id"` + ProjectID int64 `json:"project_id"` + InstanceID int64 `json:"instance_id"` + StageKey string `json:"stage_key"` + DocTasks []string `json:"doc_task_keys"` + Reason string `json:"reason"` + RequestedBy int64 `json:"requested_by"` + Status string `json:"status"` +} + +// RecheckMark is a soft signal on a completed stage. +type RecheckMark struct { + ID int64 `json:"id"` + ProjectID int64 `json:"project_id"` + StageKey string `json:"stage_key"` + SourceDoc string `json:"source_doc_id"` + Status string `json:"status"` + Reason string `json:"reason"` + AckedBy int64 `json:"acked_by,omitempty"` +} diff --git a/internal/traceimpact/walk.go b/internal/traceimpact/walk.go new file mode 100644 index 00000000..9c691c41 --- /dev/null +++ b/internal/traceimpact/walk.go @@ -0,0 +1,59 @@ +package traceimpact + +// WalkHit is one node reached by a reverse walk. +type WalkHit struct { + ItemID int64 + EdgeID int64 + Depth int + Via string +} + +// WalkReverse follows confirmed edges from changed items to their dependents. +// An edge From --depends_on/refines--> To means From depends on To. +// Changing To therefore impacts From. Candidate edges are not followed. +func WalkReverse(edges []Edge, start []int64, maxDepth, maxNodes int) (hits []WalkHit, truncated bool) { + if maxDepth <= 0 { + maxDepth = MaxWalkDepth + } + if maxNodes <= 0 { + maxNodes = MaxWalkNodes + } + startSet := map[int64]bool{} + for _, id := range start { + startSet[id] = true + } + type node struct { + id int64 + depth int + } + queue := make([]node, 0, len(start)) + for _, id := range start { + queue = append(queue, node{id: id, depth: 0}) + } + seen := map[int64]bool{} + for len(queue) > 0 { + cur := queue[0] + queue = queue[1:] + if seen[cur.id] { + continue + } + seen[cur.id] = true + if len(seen) > maxNodes { + return hits, true + } + if cur.depth >= maxDepth { + continue + } + for _, e := range edges { + if e.ReviewState != ReviewConfirmed || e.ToItemID != cur.id { + continue + } + if startSet[e.FromItemID] || seen[e.FromItemID] { + continue + } + hits = append(hits, WalkHit{ItemID: e.FromItemID, EdgeID: e.ID, Depth: cur.depth + 1, Via: e.RelationType}) + queue = append(queue, node{id: e.FromItemID, depth: cur.depth + 1}) + } + } + return hits, false +}