feat(knowledge): W3.2 line-range chunks + W4.5 drift check
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) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.7
@@ -52,3 +52,21 @@ properties:
|
||||
type: array
|
||||
items:
|
||||
type: string
|
||||
source_start_line:
|
||||
description: |
|
||||
1-indexed first line of the chunk in the normalized markdown
|
||||
content. Optional for backwards compatibility with pre-W3.2
|
||||
chunks; required for new chunks.
|
||||
type:
|
||||
- integer
|
||||
- "null"
|
||||
minimum: 1
|
||||
source_end_line:
|
||||
description: |
|
||||
1-indexed last line of the chunk (inclusive). Pair with
|
||||
`source_start_line` to map an agent citation `path:Lx-Ly` back
|
||||
to the chunk that covers those lines.
|
||||
type:
|
||||
- integer
|
||||
- "null"
|
||||
minimum: 1
|
||||
|
||||
+90
-11
@@ -394,34 +394,113 @@ def extract_markdown_title(content: str, source_path: str, strategy: str) -> str
|
||||
return Path(source_path).stem.replace("-", " ").replace("_", " ").strip()
|
||||
|
||||
|
||||
def chunk_markdown(
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class MarkdownChunk:
|
||||
"""A chunk plus the 1-indexed line range from the source markdown.
|
||||
|
||||
Used by ingest to surface `source_start_line` / `source_end_line` on
|
||||
every chunk so downstream validators (validate_citations) can
|
||||
resolve `path:Lx-Ly` citations to actual chunk content rather than
|
||||
just bounds-check the file length.
|
||||
"""
|
||||
|
||||
text: str
|
||||
start_line: int # 1-indexed inclusive
|
||||
end_line: int # 1-indexed inclusive
|
||||
|
||||
|
||||
def chunk_markdown_with_lines(
|
||||
content: str, max_characters: int = DEFAULT_MAX_CHUNK_CHARACTERS
|
||||
) -> list[str]:
|
||||
"""Chunk markdown into paragraph groups."""
|
||||
) -> list[MarkdownChunk]:
|
||||
"""Chunk markdown into paragraph groups, preserving source line ranges.
|
||||
|
||||
Splits on blank-line paragraph boundaries; groups paragraphs into
|
||||
chunks bounded by `max_characters`; tracks the first and last source
|
||||
line number of each chunk. Line numbers count blank-line separators
|
||||
in the original content (so chunk N's start_line and chunk N-1's
|
||||
end_line never overlap).
|
||||
|
||||
Returns chunks in document order. Empty content yields [].
|
||||
"""
|
||||
|
||||
# Walk lines, accumulate paragraphs with line ranges. A paragraph
|
||||
# ends at a blank line or EOF.
|
||||
paragraphs: list[tuple[str, int, int]] = []
|
||||
current_lines: list[str] = []
|
||||
current_start = -1
|
||||
line_no = 0
|
||||
for raw_line in content.split("\n"):
|
||||
line_no += 1
|
||||
if raw_line.strip():
|
||||
if not current_lines:
|
||||
current_start = line_no
|
||||
current_lines.append(raw_line)
|
||||
else:
|
||||
if current_lines:
|
||||
paragraphs.append((
|
||||
"\n".join(current_lines).strip(),
|
||||
current_start,
|
||||
line_no - 1,
|
||||
))
|
||||
current_lines = []
|
||||
current_start = -1
|
||||
if current_lines:
|
||||
paragraphs.append((
|
||||
"\n".join(current_lines).strip(),
|
||||
current_start,
|
||||
line_no,
|
||||
))
|
||||
|
||||
paragraphs = [item.strip() for item in content.split("\n\n") if item.strip()]
|
||||
if not paragraphs:
|
||||
return []
|
||||
|
||||
chunks: list[str] = []
|
||||
chunks: list[MarkdownChunk] = []
|
||||
current_parts: list[str] = []
|
||||
current_length = 0
|
||||
for paragraph in paragraphs:
|
||||
paragraph_length = len(paragraph)
|
||||
current_chunk_start = -1
|
||||
current_chunk_end = -1
|
||||
for text, start, end in paragraphs:
|
||||
paragraph_length = len(text)
|
||||
if current_parts and current_length + paragraph_length > max_characters:
|
||||
chunks.append("\n\n".join(current_parts))
|
||||
current_parts = [paragraph]
|
||||
chunks.append(MarkdownChunk(
|
||||
text="\n\n".join(current_parts),
|
||||
start_line=current_chunk_start,
|
||||
end_line=current_chunk_end,
|
||||
))
|
||||
current_parts = [text]
|
||||
current_length = paragraph_length
|
||||
current_chunk_start = start
|
||||
current_chunk_end = end
|
||||
continue
|
||||
current_parts.append(paragraph)
|
||||
if not current_parts:
|
||||
current_chunk_start = start
|
||||
current_parts.append(text)
|
||||
current_length += paragraph_length
|
||||
current_chunk_end = end
|
||||
|
||||
if current_parts:
|
||||
chunks.append("\n\n".join(current_parts))
|
||||
chunks.append(MarkdownChunk(
|
||||
text="\n\n".join(current_parts),
|
||||
start_line=current_chunk_start,
|
||||
end_line=current_chunk_end,
|
||||
))
|
||||
|
||||
return chunks
|
||||
|
||||
|
||||
def chunk_markdown(
|
||||
content: str, max_characters: int = DEFAULT_MAX_CHUNK_CHARACTERS
|
||||
) -> list[str]:
|
||||
"""Backwards-compatible string-only chunker.
|
||||
|
||||
Kept for callers that don't need line ranges. New code should call
|
||||
`chunk_markdown_with_lines` and read `MarkdownChunk.text` plus
|
||||
`start_line` / `end_line`.
|
||||
"""
|
||||
|
||||
return [c.text for c in chunk_markdown_with_lines(content, max_characters)]
|
||||
|
||||
|
||||
def knowledge_paths(deploy_root: Path, environment: str) -> KnowledgePaths:
|
||||
"""Resolve the filesystem layout used by central knowledge state."""
|
||||
|
||||
|
||||
@@ -266,6 +266,8 @@ class PackSection:
|
||||
text: str
|
||||
origin: str # "module-map" | "dify" | "task-keyword"
|
||||
score: float = 0.0
|
||||
source_start_line: int | None = None # W3.2: covered line range in
|
||||
source_end_line: int | None = None # the working-tree file
|
||||
|
||||
@property
|
||||
def token_estimate(self) -> int:
|
||||
@@ -301,6 +303,12 @@ def _build_module_sections(
|
||||
doc = store.by_source_path(inputs.repo, source_path)
|
||||
if doc is None or not doc.is_retrievable():
|
||||
continue
|
||||
# Module-map sections cover the whole doc; line range is
|
||||
# 1..len(file). validate_citations uses this to attach
|
||||
# SSOT tier to file_line citations that land inside the
|
||||
# spec/ADR.
|
||||
text = _read_content(doc)
|
||||
end_line = text.count("\n") + (1 if text and not text.endswith("\n") else 0)
|
||||
sections.append(
|
||||
PackSection(
|
||||
section_id=f"{module.name}:{Path(source_path).name}",
|
||||
@@ -311,9 +319,11 @@ def _build_module_sections(
|
||||
source_path=doc.source_path,
|
||||
source_ref=doc.source_ref,
|
||||
chunk_id=None,
|
||||
text=_read_content(doc),
|
||||
text=text,
|
||||
origin="module-map",
|
||||
score=10.0,
|
||||
source_start_line=1 if text else None,
|
||||
source_end_line=end_line if end_line > 0 else None,
|
||||
)
|
||||
)
|
||||
return sections
|
||||
@@ -586,6 +596,8 @@ def _format_manifest(
|
||||
"score": s.score,
|
||||
"token_estimate": s.token_estimate,
|
||||
"text_sha256": sha256_text(s.text),
|
||||
"source_start_line": s.source_start_line,
|
||||
"source_end_line": s.source_end_line,
|
||||
}
|
||||
for s in sections
|
||||
],
|
||||
|
||||
@@ -0,0 +1,252 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Check drift between git publish-records and live Dify state (W4.5).
|
||||
|
||||
For each published record, hit Dify and verify:
|
||||
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 the recorded document_name
|
||||
|
||||
Cases caught:
|
||||
* Dify document hand-deleted via the web console
|
||||
* Dataset hand-renamed or deleted
|
||||
* remote_document_id stale (record points at gone document)
|
||||
* Records that point at a different dataset than central
|
||||
publish-targets.yaml currently maps the visibility to
|
||||
|
||||
Does NOT modify anything. Output is a per-record verdict + a summary
|
||||
suitable for CI alert threshold checks. Pair with reconcile_records.py
|
||||
to actually re-push missing chunks back into Dify.
|
||||
|
||||
CLI:
|
||||
python3 -m scripts.knowledge.dify_drift_check \\
|
||||
--deploy-root . --env dev \\
|
||||
--dify-base-url https://dify.omniverseai.net/v1 \\
|
||||
--dify-api-key "$(cat ~/.config/ljwx-dify/api-key)" \\
|
||||
--output drift-report.json
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from dataclasses import dataclass, asdict
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
||||
from .common import (
|
||||
KnowledgeContractError,
|
||||
knowledge_paths,
|
||||
load_publish_records,
|
||||
require_non_empty_str,
|
||||
)
|
||||
from .dify_client import DifyClient, DifyClientError
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class TargetVerdict:
|
||||
target_id: str
|
||||
dataset_id: str
|
||||
dataset_name: str
|
||||
remote_document_id: str | None
|
||||
drift: list[str]
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class RecordVerdict:
|
||||
document_id: str
|
||||
source_repo: str
|
||||
source_path: str
|
||||
record_status: str
|
||||
targets: list[TargetVerdict]
|
||||
|
||||
@property
|
||||
def is_clean(self) -> bool:
|
||||
return all(not t.drift for t in self.targets)
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description="Check drift between git publish-records and Dify")
|
||||
parser.add_argument("--deploy-root", type=Path, required=True)
|
||||
parser.add_argument("--env", default="dev")
|
||||
parser.add_argument("--source-repo", help="filter to one owner/repo")
|
||||
parser.add_argument("--dify-base-url", required=True)
|
||||
parser.add_argument("--dify-api-key", required=True)
|
||||
parser.add_argument(
|
||||
"--output",
|
||||
type=Path,
|
||||
help="optional JSON report path; default stdout",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--limit",
|
||||
type=int,
|
||||
default=0,
|
||||
help="check at most N records (0 = no limit); useful for fast smoke",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--fail-on-drift",
|
||||
action="store_true",
|
||||
help="exit non-zero if any drift found (default: report only)",
|
||||
)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = parse_args()
|
||||
deploy_root = args.deploy_root.resolve()
|
||||
paths = knowledge_paths(deploy_root, args.env)
|
||||
records = load_publish_records(paths.publish_records_root)
|
||||
|
||||
try:
|
||||
client = DifyClient(
|
||||
base_url=require_non_empty_str(args.dify_base_url, "--dify-base-url"),
|
||||
api_key=require_non_empty_str(args.dify_api_key, "--dify-api-key"),
|
||||
)
|
||||
except DifyClientError as exc:
|
||||
print(f"ERROR: {exc}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
# Cache datasets list once; Dify has ~5 datasets, ~hundreds of
|
||||
# documents each, so we batch lookups per dataset.
|
||||
try:
|
||||
live_datasets = {ds.dataset_id: ds.name for ds in client.list_datasets()}
|
||||
except DifyClientError as exc:
|
||||
print(f"ERROR listing datasets: {exc}", file=sys.stderr)
|
||||
return 2
|
||||
live_dataset_docs: dict[str, dict[str, str]] = {} # dataset_id -> {doc_id -> doc_name}
|
||||
|
||||
def docs_for(dataset_id: str) -> dict[str, str]:
|
||||
if dataset_id not in live_dataset_docs:
|
||||
try:
|
||||
live_dataset_docs[dataset_id] = {
|
||||
d.document_id: d.name for d in client.list_documents(dataset_id)
|
||||
}
|
||||
except DifyClientError as exc:
|
||||
# Treat as empty so we record the target as drifted
|
||||
# rather than abort the whole run.
|
||||
print(
|
||||
f" WARN listing documents for {dataset_id}: {exc}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
live_dataset_docs[dataset_id] = {}
|
||||
return live_dataset_docs[dataset_id]
|
||||
|
||||
verdicts: list[RecordVerdict] = []
|
||||
checked = 0
|
||||
for document_id, record in records.items():
|
||||
source_repo = str(record.get("source_repo", ""))
|
||||
if args.source_repo and source_repo != args.source_repo:
|
||||
continue
|
||||
if args.limit and checked >= args.limit:
|
||||
break
|
||||
checked += 1
|
||||
|
||||
status = str(record.get("status", ""))
|
||||
if status != "published":
|
||||
# invalidated records intentionally have no Dify presence;
|
||||
# skip from drift check (reconcile_records owns those).
|
||||
continue
|
||||
|
||||
targets_raw = record.get("targets") or []
|
||||
target_verdicts: list[TargetVerdict] = []
|
||||
for t in targets_raw:
|
||||
if not isinstance(t, dict):
|
||||
continue
|
||||
target_id = str(t.get("target_id", ""))
|
||||
dataset_id = str(t.get("dataset_id", ""))
|
||||
dataset_name = str(t.get("dataset_name", ""))
|
||||
remote_document_id = t.get("remote_document_id")
|
||||
remote_document_id = (
|
||||
str(remote_document_id) if isinstance(remote_document_id, str) else None
|
||||
)
|
||||
document_name = str(t.get("document_name", ""))
|
||||
|
||||
drift: list[str] = []
|
||||
if not dataset_id:
|
||||
drift.append("missing-dataset-id")
|
||||
elif dataset_id not in live_datasets:
|
||||
drift.append(f"dataset-deleted: {dataset_id}")
|
||||
else:
|
||||
live_name = live_datasets[dataset_id]
|
||||
if dataset_name and live_name != dataset_name:
|
||||
drift.append(
|
||||
f"dataset-renamed: recorded={dataset_name} live={live_name}"
|
||||
)
|
||||
docs = docs_for(dataset_id)
|
||||
if remote_document_id and remote_document_id not in docs:
|
||||
drift.append(
|
||||
f"document-missing-or-renamed: {remote_document_id}"
|
||||
)
|
||||
elif remote_document_id and document_name:
|
||||
live_doc_name = docs[remote_document_id]
|
||||
if live_doc_name != document_name:
|
||||
drift.append(
|
||||
f"document-renamed: recorded={document_name} live={live_doc_name}"
|
||||
)
|
||||
|
||||
target_verdicts.append(
|
||||
TargetVerdict(
|
||||
target_id=target_id,
|
||||
dataset_id=dataset_id,
|
||||
dataset_name=dataset_name,
|
||||
remote_document_id=remote_document_id,
|
||||
drift=drift,
|
||||
)
|
||||
)
|
||||
|
||||
verdicts.append(
|
||||
RecordVerdict(
|
||||
document_id=document_id,
|
||||
source_repo=source_repo,
|
||||
source_path=str(record.get("source_path", "")),
|
||||
record_status=status,
|
||||
targets=target_verdicts,
|
||||
)
|
||||
)
|
||||
|
||||
drifted = [v for v in verdicts if not v.is_clean]
|
||||
report = {
|
||||
"env": args.env,
|
||||
"source_repo_filter": args.source_repo,
|
||||
"records_checked": len(verdicts),
|
||||
"records_drifted": len(drifted),
|
||||
"live_datasets": [{"id": k, "name": v} for k, v in live_datasets.items()],
|
||||
"drifted_records": [
|
||||
{
|
||||
"document_id": v.document_id,
|
||||
"source_repo": v.source_repo,
|
||||
"source_path": v.source_path,
|
||||
"targets": [asdict(t) for t in v.targets if t.drift],
|
||||
}
|
||||
for v in drifted
|
||||
],
|
||||
"summary_by_repo": {},
|
||||
}
|
||||
# Per-repo summary so CI can route alerts.
|
||||
from collections import Counter
|
||||
|
||||
repo_total = Counter(v.source_repo for v in verdicts)
|
||||
repo_drift = Counter(v.source_repo for v in drifted)
|
||||
report["summary_by_repo"] = {
|
||||
repo: {"checked": repo_total[repo], "drifted": repo_drift.get(repo, 0)}
|
||||
for repo in sorted(repo_total.keys())
|
||||
}
|
||||
|
||||
body = json.dumps(report, ensure_ascii=False, indent=2) + "\n"
|
||||
if args.output is not None:
|
||||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.output.write_text(body, encoding="utf-8")
|
||||
else:
|
||||
sys.stdout.write(body)
|
||||
|
||||
print(
|
||||
f"\ndify_drift_check: {len(drifted)} drift / {len(verdicts)} checked",
|
||||
file=sys.stderr,
|
||||
)
|
||||
if drifted and args.fail_on_drift:
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -25,6 +25,7 @@ from .common import (
|
||||
build_event,
|
||||
cast_mapping_list,
|
||||
chunk_markdown,
|
||||
chunk_markdown_with_lines,
|
||||
event_path_for_record,
|
||||
knowledge_paths,
|
||||
load_acl_policies,
|
||||
@@ -300,7 +301,7 @@ def build_chunk_rows(
|
||||
normalized_document.get("acl_expression"), "normalized.acl_expression"
|
||||
)
|
||||
chunks: list[JSONMapping] = []
|
||||
for index, chunk_text in enumerate(chunk_markdown(content)):
|
||||
for index, chunk in enumerate(chunk_markdown_with_lines(content)):
|
||||
chunks.append(
|
||||
{
|
||||
"chunk_id": f"{document_id}:chunk:{index:04d}",
|
||||
@@ -310,10 +311,14 @@ def build_chunk_rows(
|
||||
"visibility": visibility,
|
||||
"acl_expression": acl_expression,
|
||||
"customer_id": None,
|
||||
"token_count": max(1, len(chunk_text.strip()) // 4),
|
||||
"token_count": max(1, len(chunk.text.strip()) // 4),
|
||||
"embedding_model": None,
|
||||
"text_hash": sha256_text(chunk_text),
|
||||
"text_hash": sha256_text(chunk.text),
|
||||
"published_targets": [],
|
||||
# W3.2: source line range so validate_citations can
|
||||
# resolve `path:Lx-Ly` to the actual chunk.
|
||||
"source_start_line": chunk.start_line,
|
||||
"source_end_line": chunk.end_line,
|
||||
}
|
||||
)
|
||||
return chunks
|
||||
|
||||
@@ -220,9 +220,6 @@ def verify(
|
||||
severity_cap="info",
|
||||
reason="file does not exist in working tree",
|
||||
)
|
||||
# Verify line range fits the file. Line content match needs per-chunk
|
||||
# ranges in the pack (W3.2); for now we only check the range is in
|
||||
# bounds.
|
||||
try:
|
||||
line_count = sum(1 for _ in path.open("r", encoding="utf-8"))
|
||||
except OSError as exc:
|
||||
@@ -234,7 +231,8 @@ def verify(
|
||||
reason=f"file read error: {exc}",
|
||||
)
|
||||
end = citation.line_end or citation.line_start or 0
|
||||
if (citation.line_start or 0) < 1 or end > line_count:
|
||||
start = citation.line_start or 0
|
||||
if start < 1 or end > line_count:
|
||||
return Verdict(
|
||||
citation=citation,
|
||||
verified=False,
|
||||
@@ -242,8 +240,39 @@ def verify(
|
||||
severity_cap="info",
|
||||
reason=f"line range out of bounds (file has {line_count} lines)",
|
||||
)
|
||||
# file_line citations cannot establish tier; treat as reference until
|
||||
# W3.2 maps line ranges back to a pack chunk.
|
||||
|
||||
# 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,
|
||||
|
||||
Reference in New Issue
Block a user