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 <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-20 15:35:07 +00:00
co-authored by Cursor
parent e3a5b2851e
commit 33ee59641f
11 changed files with 1808 additions and 0 deletions
@@ -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;
+74
View File
@@ -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
}
+27
View File
@@ -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")
}
}
+28
View File
@@ -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))
}
+307
View File
@@ -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
}
+114
View File
@@ -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
}
+380
View File
@@ -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
}
+230
View File
@@ -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)
}
}
+317
View File
@@ -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
}
+164
View File
@@ -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"`
}
+59
View File
@@ -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
}