Benjamin_Boenisch/breakpilot-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(control-pipeline): add anchor backfill endpoint + normalize target_audience

Browse Source 
- Add POST /v1/canonical/generate/backfill-anchors endpoint for batch
  populating open_anchors on controls generated with skip_web_search=true
- Uses AnchorFinder Stage A (RAG search) to find OWASP/NIST/ENISA refs
- Background job with progress tracking (same pattern as other backfills)
- Promotes needs_review controls that gain anchors to draft state
- Target audience normalization (enterprise/authority/provider → JSON arrays)
  already applied via SQL

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
2026-04-21 18:04:50 +02:00
parent e5bb8e65e8
commit 91f4202e88
2 changed files with 184 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
control-pipeline/services/control_generator.py
View File

@@ -1555,14 +1555,15 @@ Gib ein JSON-Array zurueck mit GENAU {len(chunks)} Elementen. Fuer Aspekte ohne
final.append(control)
continue
# Anchor search
try:
from .anchor_finder import AnchorFinder
finder = AnchorFinder(self.rag)
anchors = await finder.find_anchors(control, skip_web=config.skip_web_search)
control.open_anchors = [asdict(a) if hasattr(a, '__dataclass_fields__') else a for a in anchors]
except Exception as e:
logger.warning("Anchor search failed: %s", e)
# Anchor search — skip entirely when skip_web_search=true (backfilled later)
if not config.skip_web_search:
try:
from .anchor_finder import AnchorFinder
finder = AnchorFinder(self.rag)
anchors = await finder.find_anchors(control, skip_web=False)
control.open_anchors = [asdict(a) if hasattr(a, '__dataclass_fields__') else a for a in anchors]
except Exception as e:
logger.warning("Anchor search failed: %s", e)
# Release state
if control.license_rule in (1, 2):
@@ -2402,14 +2403,15 @@ Kategorien: {CATEGORY_LIST_STR}"""
control.generation_metadata["similar_controls"] = duplicates
return control
# Stage 5: Anchor Search (imported from anchor_finder)
try:
from .anchor_finder import AnchorFinder
finder = AnchorFinder(self.rag)
anchors = await finder.find_anchors(control, skip_web=config.skip_web_search)
control.open_anchors = [asdict(a) if hasattr(a, '__dataclass_fields__') else a for a in anchors]
except Exception as e:
logger.warning("Anchor search failed: %s", e)
# Stage 5: Anchor Search — skip entirely when skip_web_search=true (backfilled later)
if not config.skip_web_search:
try:
from .anchor_finder import AnchorFinder
finder = AnchorFinder(self.rag)
anchors = await finder.find_anchors(control, skip_web=False)
control.open_anchors = [asdict(a) if hasattr(a, '__dataclass_fields__') else a for a in anchors]
except Exception as e:
logger.warning("Anchor search failed: %s", e)
# Determine release state
if control.license_rule in (1, 2):