brunoandClaude Opus 4.7
e400e98db4
W3.2 (source line ranges per chunk): - common.MarkdownChunk dataclass + chunk_markdown_with_lines() tracks 1-indexed source line ranges as it groups paragraphs into chunks. Old chunk_markdown() preserved as a thin wrapper for callers that don't need line ranges. - ingest_artifact.build_chunks_for_document now emits source_start_line / source_end_line on every chunk. - knowledge-chunk schema accepts the two new fields (nullable for backward compat with pre-W3.2 chunks; required for new ones). - context_pack.PackSection carries the same fields; module-map sections record start=1 / end=line_count of the full file. Manifest output and pack.md citation grammar are unchanged. - validate_citations.verify() for kind=file_line now searches the pack for sections whose source_path matches AND whose line range overlaps the citation. On hit, inherits that section's tier (typically ssot for module.yml/ADR sections). On no overlap, falls back to the existing weak verification (bounds-only). Smoke (against ljwx-platform with module-map): - spec/modules/license/module.yml:L1-L14 -> tier=ssot, severity=block - docs/adr/006-security-hardening.md:L40-L60 -> tier=ssot, severity=block - :L200-L300 (out of bounds) -> unverified, info - nonexistent/path.md -> unverified, info W4.5 (Dify drift check): - new scripts/knowledge/dify_drift_check.py: read-only sweep that for each published record verifies (1) the dataset still exists with the recorded dataset_id, (2) the document still exists with the recorded remote_document_id, (3) the document name on Dify matches. - Emits per-record + per-repo JSON report; exits non-zero with --fail-on-drift for CI. - Smoke against ljwx-platform: 0 drift in 10-record sample. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
383 lines
12 KiB
Python
383 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Verify citations in agent output against a context pack and the working
|
|
tree.
|
|
|
|
Agent findings must cite using one of three forms (see context_pack.py):
|
|
[kg:{repo}:{document_id}:{chunk_id}]
|
|
[kg:{repo}:{source_path}]
|
|
{source_path}:L{x}-L{y} or {source_path}:L{x}
|
|
|
|
For each citation we resolve and produce a triple:
|
|
verified True if the cited target actually exists.
|
|
tier ssot | proposed | reference | legacy | unknown
|
|
severity_cap block | warn | info
|
|
- unverified citations always cap at info
|
|
- tier=legacy caps at warn (cited source may be stale)
|
|
- tier=proposed caps at warn (cannot block on draft spec)
|
|
- tier=ssot/reference allows block
|
|
|
|
The script does not modify findings. It emits a per-citation verdict so a
|
|
downstream review pipeline can apply the severity cap and decide what to
|
|
surface as blocking, warning, or informational.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Final
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Citation grammar
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# [kg:repo:doc_id:chunk_id] repo may contain `/`; doc_id and chunk_id may
|
|
# contain `:` only inside chunk_id segment. We accept owner/repo with /.
|
|
KG_CHUNK_RE: Final[re.Pattern[str]] = re.compile(
|
|
r"\[kg:(?P<repo>[\w./\-]+):(?P<doc_id>[\w.\-]+):(?P<chunk_id>[\w.:\-]+)\]"
|
|
)
|
|
KG_DOC_RE: Final[re.Pattern[str]] = re.compile(
|
|
r"\[kg:(?P<repo>[\w./\-]+):(?P<source_path>[\w./\-]+)\]"
|
|
)
|
|
# path:Lx-Ly or path:Lx; require path to look like a real source file
|
|
FILE_LINE_RE: Final[re.Pattern[str]] = re.compile(
|
|
r"(?P<path>[\w./\-]+\.(?:py|ts|tsx|js|jsx|java|go|rs|md|yml|yaml|sql|toml|json|sh)):L(?P<start>\d+)(?:-L?(?P<end>\d+))?"
|
|
)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Citation:
|
|
kind: str # kg_chunk | kg_doc | file_line
|
|
text: str
|
|
repo: str | None = None
|
|
document_id: str | None = None
|
|
chunk_id: str | None = None
|
|
source_path: str | None = None
|
|
line_start: int | None = None
|
|
line_end: int | None = None
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Verdict:
|
|
citation: Citation
|
|
verified: bool
|
|
tier: str
|
|
severity_cap: str
|
|
reason: str
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Verification
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
SEVERITY_BY_TIER: Final[dict[str, str]] = {
|
|
"ssot": "block",
|
|
"reference": "block",
|
|
"proposed": "warn",
|
|
"legacy": "warn",
|
|
"unknown": "warn",
|
|
}
|
|
|
|
|
|
def extract_citations(text: str) -> list[Citation]:
|
|
"""Pull every citation from a blob of agent text.
|
|
|
|
Order: longest patterns first so kg_chunk doesn't get re-matched as
|
|
kg_doc + chunk text.
|
|
"""
|
|
citations: list[Citation] = []
|
|
# Track spans to avoid double-counting
|
|
consumed: list[tuple[int, int]] = []
|
|
|
|
def overlaps(start: int, end: int) -> bool:
|
|
return any(not (end <= cs or start >= ce) for cs, ce in consumed)
|
|
|
|
for match in KG_CHUNK_RE.finditer(text):
|
|
if overlaps(match.start(), match.end()):
|
|
continue
|
|
consumed.append((match.start(), match.end()))
|
|
citations.append(
|
|
Citation(
|
|
kind="kg_chunk",
|
|
text=match.group(0),
|
|
repo=match.group("repo"),
|
|
document_id=match.group("doc_id"),
|
|
chunk_id=match.group("chunk_id"),
|
|
)
|
|
)
|
|
|
|
for match in KG_DOC_RE.finditer(text):
|
|
if overlaps(match.start(), match.end()):
|
|
continue
|
|
# Skip if this would also match KG_CHUNK
|
|
if match.group("source_path").count(":") > 0:
|
|
continue
|
|
consumed.append((match.start(), match.end()))
|
|
citations.append(
|
|
Citation(
|
|
kind="kg_doc",
|
|
text=match.group(0),
|
|
repo=match.group("repo"),
|
|
source_path=match.group("source_path"),
|
|
)
|
|
)
|
|
|
|
for match in FILE_LINE_RE.finditer(text):
|
|
if overlaps(match.start(), match.end()):
|
|
continue
|
|
consumed.append((match.start(), match.end()))
|
|
citations.append(
|
|
Citation(
|
|
kind="file_line",
|
|
text=match.group(0),
|
|
source_path=match.group("path"),
|
|
line_start=int(match.group("start")),
|
|
line_end=int(match.group("end")) if match.group("end") else None,
|
|
)
|
|
)
|
|
|
|
return citations
|
|
|
|
|
|
def _load_pack(pack_path: Path) -> dict:
|
|
return json.loads(pack_path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def _index_pack(pack: dict) -> tuple[dict[tuple[str, str], dict], dict[tuple[str, str], dict]]:
|
|
by_chunk: dict[tuple[str, str], dict] = {}
|
|
by_doc: dict[tuple[str, str], dict] = {}
|
|
for section in pack.get("sections", []) or []:
|
|
repo = str(section.get("source_repo", ""))
|
|
doc_id = str(section.get("document_id", ""))
|
|
source_path = str(section.get("source_path", ""))
|
|
chunk_id = section.get("chunk_id")
|
|
if chunk_id:
|
|
by_chunk[(repo, str(chunk_id))] = section
|
|
by_doc[(repo, source_path)] = section
|
|
by_doc[(repo, doc_id)] = section
|
|
return by_chunk, by_doc
|
|
|
|
|
|
def verify(
|
|
citation: Citation,
|
|
pack_by_chunk: dict[tuple[str, str], dict],
|
|
pack_by_doc: dict[tuple[str, str], dict],
|
|
repo_root: Path,
|
|
) -> Verdict:
|
|
if citation.kind == "kg_chunk":
|
|
key = (citation.repo or "", citation.chunk_id or "")
|
|
section = pack_by_chunk.get(key)
|
|
if section is None:
|
|
return Verdict(
|
|
citation=citation,
|
|
verified=False,
|
|
tier="unknown",
|
|
severity_cap="info",
|
|
reason="chunk_id not present in pack",
|
|
)
|
|
tier = str(section.get("tier", "unknown"))
|
|
return Verdict(
|
|
citation=citation,
|
|
verified=True,
|
|
tier=tier,
|
|
severity_cap=SEVERITY_BY_TIER.get(tier, "warn"),
|
|
reason="chunk resolved",
|
|
)
|
|
|
|
if citation.kind == "kg_doc":
|
|
key = (citation.repo or "", citation.source_path or "")
|
|
section = pack_by_doc.get(key)
|
|
if section is None:
|
|
return Verdict(
|
|
citation=citation,
|
|
verified=False,
|
|
tier="unknown",
|
|
severity_cap="info",
|
|
reason="source_path not present in pack",
|
|
)
|
|
tier = str(section.get("tier", "unknown"))
|
|
return Verdict(
|
|
citation=citation,
|
|
verified=True,
|
|
tier=tier,
|
|
severity_cap=SEVERITY_BY_TIER.get(tier, "warn"),
|
|
reason="document resolved",
|
|
)
|
|
|
|
if citation.kind == "file_line":
|
|
path = repo_root / (citation.source_path or "")
|
|
if not path.exists():
|
|
return Verdict(
|
|
citation=citation,
|
|
verified=False,
|
|
tier="unknown",
|
|
severity_cap="info",
|
|
reason="file does not exist in working tree",
|
|
)
|
|
try:
|
|
line_count = sum(1 for _ in path.open("r", encoding="utf-8"))
|
|
except OSError as exc:
|
|
return Verdict(
|
|
citation=citation,
|
|
verified=False,
|
|
tier="unknown",
|
|
severity_cap="info",
|
|
reason=f"file read error: {exc}",
|
|
)
|
|
end = citation.line_end or citation.line_start or 0
|
|
start = citation.line_start or 0
|
|
if start < 1 or end > line_count:
|
|
return Verdict(
|
|
citation=citation,
|
|
verified=False,
|
|
tier="unknown",
|
|
severity_cap="info",
|
|
reason=f"line range out of bounds (file has {line_count} lines)",
|
|
)
|
|
|
|
# W3.2: try to match the cited line range to a pack chunk so we
|
|
# can adopt that chunk's tier. Pack sections may carry
|
|
# source_start_line / source_end_line per chunk (added in the
|
|
# W3.2 schema bump); fall back to weak-verification if none of
|
|
# the pack chunks for this source_path covers the cited range.
|
|
matching_section = None
|
|
for section in pack_by_doc.values():
|
|
if section.get("source_path") != citation.source_path:
|
|
continue
|
|
chunk_start = section.get("source_start_line")
|
|
chunk_end = section.get("source_end_line")
|
|
if not isinstance(chunk_start, int) or not isinstance(chunk_end, int):
|
|
continue
|
|
# range overlap test
|
|
if start <= chunk_end and end >= chunk_start:
|
|
matching_section = section
|
|
break
|
|
|
|
if matching_section is not None:
|
|
tier = str(matching_section.get("tier", "reference"))
|
|
return Verdict(
|
|
citation=citation,
|
|
verified=True,
|
|
tier=tier,
|
|
severity_cap=SEVERITY_BY_TIER.get(tier, "warn"),
|
|
reason=(
|
|
f"file:line in bounds + overlaps pack chunk "
|
|
f"({matching_section.get('source_start_line')}-"
|
|
f"{matching_section.get('source_end_line')})"
|
|
),
|
|
)
|
|
# No chunk match → weak verification (bounds-only, tier unknown)
|
|
return Verdict(
|
|
citation=citation,
|
|
verified=True,
|
|
tier="reference",
|
|
severity_cap="warn",
|
|
reason="file:line in bounds (weak verification until W3.2)",
|
|
)
|
|
|
|
return Verdict(
|
|
citation=citation,
|
|
verified=False,
|
|
tier="unknown",
|
|
severity_cap="info",
|
|
reason="unsupported citation kind",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _read_findings(path: Path) -> str:
|
|
if path == Path("-"):
|
|
return sys.stdin.read()
|
|
return path.read_text(encoding="utf-8")
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Validate citations in agent findings against a context pack"
|
|
)
|
|
parser.add_argument(
|
|
"--pack", type=Path, required=True, help="context-pack.json"
|
|
)
|
|
parser.add_argument(
|
|
"--findings",
|
|
type=Path,
|
|
required=True,
|
|
help="agent output as text or json; use '-' for stdin",
|
|
)
|
|
parser.add_argument(
|
|
"--repo-root",
|
|
type=Path,
|
|
default=Path.cwd(),
|
|
help="working tree for file_line resolution",
|
|
)
|
|
parser.add_argument(
|
|
"--output",
|
|
type=Path,
|
|
default=None,
|
|
help="optional: write verdict json; default is stdout",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
pack = _load_pack(args.pack)
|
|
by_chunk, by_doc = _index_pack(pack)
|
|
|
|
text = _read_findings(args.findings)
|
|
citations = extract_citations(text)
|
|
verdicts = [verify(c, by_chunk, by_doc, args.repo_root) for c in citations]
|
|
|
|
summary = {
|
|
"pack": str(args.pack),
|
|
"total": len(verdicts),
|
|
"verified": sum(1 for v in verdicts if v.verified),
|
|
"by_severity": {
|
|
"block": sum(1 for v in verdicts if v.severity_cap == "block"),
|
|
"warn": sum(1 for v in verdicts if v.severity_cap == "warn"),
|
|
"info": sum(1 for v in verdicts if v.severity_cap == "info"),
|
|
},
|
|
"by_tier": {
|
|
tier: sum(1 for v in verdicts if v.tier == tier)
|
|
for tier in ("ssot", "reference", "proposed", "legacy", "unknown")
|
|
},
|
|
"verdicts": [
|
|
{
|
|
"citation_text": v.citation.text,
|
|
"kind": v.citation.kind,
|
|
"repo": v.citation.repo,
|
|
"document_id": v.citation.document_id,
|
|
"chunk_id": v.citation.chunk_id,
|
|
"source_path": v.citation.source_path,
|
|
"line_start": v.citation.line_start,
|
|
"line_end": v.citation.line_end,
|
|
"verified": v.verified,
|
|
"tier": v.tier,
|
|
"severity_cap": v.severity_cap,
|
|
"reason": v.reason,
|
|
}
|
|
for v in verdicts
|
|
],
|
|
}
|
|
out = json.dumps(summary, ensure_ascii=False, indent=2)
|
|
if args.output is not None:
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(out + "\n", encoding="utf-8")
|
|
else:
|
|
print(out)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|