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]>
861 lines
30 KiB
Python
861 lines
30 KiB
Python
#!/usr/bin/env python3
|
|
"""Build a per-task knowledge context pack for Codex review and Claude
|
|
implementation.
|
|
|
|
Pipeline (high level):
|
|
inputs (repo + mode + task-text and/or changed-files)
|
|
-> module-map resolution (changed-files -> modules -> spec/ADR paths)
|
|
-> local publish-record lookup (read content_ref markdown locally)
|
|
-> optional Dify retrieval supplement (if KNOWLEDGE_DIFY_* env present)
|
|
-> token-bounded composition (SSOT first, then ADR, then supplement)
|
|
-> output: context-pack.md (for agent consumption) + context-pack.json
|
|
(machine manifest with chunk_id, source_ref, tier metadata)
|
|
|
|
Citation grammar (consumed by validate_citations.py):
|
|
[kg:{repo}:{document_id}:{chunk_id}] chunk-level, fully verifiable
|
|
[kg:{repo}:{source_path}] document-level, path-verifiable
|
|
{source_path}:L{x}-L{y} source-tree, file existence checked
|
|
(line range will be verifiable once
|
|
W3.2 lands per-chunk line ranges)
|
|
|
|
Operates without Dify creds — falls back to local-content scoring so dev
|
|
loops work offline. Provide KNOWLEDGE_DIFY_BASE_URL + KNOWLEDGE_DIFY_API_KEY
|
|
to enable live retrieval.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import fnmatch
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Final
|
|
|
|
from .common import (
|
|
KnowledgeContractError,
|
|
load_json_mapping,
|
|
load_yaml_mapping,
|
|
sha256_text,
|
|
utc_now,
|
|
write_json,
|
|
)
|
|
from .dify_client import DifyClient, DifyClientError, DifyRetrievalHit
|
|
|
|
|
|
DEFAULT_TOKEN_BUDGET: Final[int] = 20_000
|
|
DEFAULT_TOP_K: Final[int] = 12
|
|
# Conservative ratio for mixed Chinese/English markdown.
|
|
CHARS_PER_TOKEN: Final[float] = 3.5
|
|
|
|
TIER_ORDER: Final[dict[str, int]] = {
|
|
"ssot": 0,
|
|
"proposed": 1,
|
|
"reference": 2,
|
|
"legacy": 3,
|
|
"unknown": 4,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Inputs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PackInputs:
|
|
repo: str
|
|
mode: str
|
|
task_text: str | None
|
|
changed_files: tuple[str, ...]
|
|
repo_root: Path
|
|
deploy_root: Path
|
|
output_dir: Path
|
|
env: str
|
|
token_budget: int
|
|
top_k: int
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Module map
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ModuleEntry:
|
|
name: str
|
|
paths: tuple[str, ...]
|
|
excluded_paths: tuple[str, ...]
|
|
spec: str | None
|
|
adrs: tuple[str, ...]
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ModuleMap:
|
|
modules: tuple[ModuleEntry, ...]
|
|
fallback_adrs: tuple[str, ...]
|
|
|
|
def resolve(self, changed_path: str) -> list[ModuleEntry]:
|
|
"""Return modules whose paths match, longest glob wins on conflict."""
|
|
candidates: list[tuple[int, ModuleEntry]] = []
|
|
for module in self.modules:
|
|
if any(fnmatch.fnmatch(changed_path, p) for p in module.excluded_paths):
|
|
continue
|
|
matched = [p for p in module.paths if fnmatch.fnmatch(changed_path, p)]
|
|
if matched:
|
|
best = max(len(p) for p in matched)
|
|
candidates.append((best, module))
|
|
if not candidates:
|
|
return []
|
|
max_len = max(score for score, _ in candidates)
|
|
return [m for score, m in candidates if score == max_len]
|
|
|
|
|
|
def load_module_map(repo_root: Path) -> ModuleMap | None:
|
|
path = repo_root / "knowledge.module-map.yaml"
|
|
if not path.exists():
|
|
return None
|
|
raw = load_yaml_mapping(path)
|
|
raw_modules = raw.get("modules", {})
|
|
if not isinstance(raw_modules, dict):
|
|
raise KnowledgeContractError(
|
|
f"{path}: modules must be a mapping"
|
|
)
|
|
modules: list[ModuleEntry] = []
|
|
for name, entry in raw_modules.items():
|
|
if not isinstance(entry, dict):
|
|
raise KnowledgeContractError(
|
|
f"{path}: modules.{name} must be a mapping"
|
|
)
|
|
modules.append(
|
|
ModuleEntry(
|
|
name=str(name),
|
|
paths=tuple(str(p) for p in entry.get("paths", []) or []),
|
|
excluded_paths=tuple(
|
|
str(p) for p in entry.get("excluded_paths", []) or []
|
|
),
|
|
spec=(str(entry["spec"]) if entry.get("spec") else None),
|
|
adrs=tuple(str(a) for a in entry.get("adrs", []) or []),
|
|
)
|
|
)
|
|
fallback = tuple(str(a) for a in raw.get("fallback_adrs", []) or [])
|
|
return ModuleMap(modules=tuple(modules), fallback_adrs=fallback)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Publish-record store (local artifacts)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PublishedDocument:
|
|
document_id: str
|
|
source_repo: str
|
|
source_id: str
|
|
source_path: str
|
|
source_ref: str | None
|
|
title: str
|
|
visibility: str
|
|
tier: str
|
|
audience: tuple[str, ...]
|
|
doc_type: str
|
|
tags: tuple[str, ...]
|
|
content_ref: Path
|
|
chunk_ref: Path
|
|
artifact_ref: Path
|
|
review_status: str
|
|
excluded_from_retrieval: bool
|
|
|
|
def is_retrievable(self) -> bool:
|
|
return (
|
|
not self.excluded_from_retrieval
|
|
and self.review_status == "approved"
|
|
)
|
|
|
|
|
|
class PublishRecordStore:
|
|
"""Indexed view of `ljwx-deploy/knowledge/publish-records/{env}/{owner__repo}/`."""
|
|
|
|
def __init__(self, deploy_root: Path, env: str) -> None:
|
|
self._deploy_root = deploy_root
|
|
self._env = env
|
|
self._by_doc_id: dict[str, PublishedDocument] = {}
|
|
self._by_source_path: dict[tuple[str, str], PublishedDocument] = {}
|
|
self._by_repo: dict[str, list[PublishedDocument]] = {}
|
|
|
|
def load_for_repo(self, repo_qualifier: str) -> int:
|
|
"""Load all publish records under one `owner__repo` directory.
|
|
|
|
`repo_qualifier` example: `lingjingwanxiang__ljwx-platform`.
|
|
Returns count loaded.
|
|
"""
|
|
base = self._deploy_root / "knowledge" / "publish-records" / self._env / repo_qualifier
|
|
if not base.exists():
|
|
return 0
|
|
loaded = 0
|
|
for record_path in sorted(base.glob("*.json")):
|
|
try:
|
|
raw = load_json_mapping(record_path)
|
|
except Exception:
|
|
continue
|
|
if raw.get("status") != "published":
|
|
continue
|
|
content_ref_raw = raw.get("content_ref")
|
|
chunk_ref_raw = raw.get("chunk_ref")
|
|
artifact_ref_raw = raw.get("artifact_ref")
|
|
if not (
|
|
isinstance(content_ref_raw, str)
|
|
and isinstance(chunk_ref_raw, str)
|
|
and isinstance(artifact_ref_raw, str)
|
|
):
|
|
continue
|
|
doc = PublishedDocument(
|
|
document_id=str(raw.get("document_id", "")),
|
|
source_repo=str(raw.get("source_repo", "")),
|
|
source_id=str(raw.get("source_id", "")),
|
|
source_path=str(raw.get("source_path", "")),
|
|
source_ref=raw.get("source_ref") if isinstance(raw.get("source_ref"), str) else None,
|
|
title=str(raw.get("title", "")),
|
|
visibility=str(raw.get("visibility", "internal")),
|
|
tier=str(raw.get("tier", "unknown")) if raw.get("tier") else "unknown",
|
|
audience=tuple(
|
|
str(a) for a in (raw.get("audience") or []) if isinstance(a, str)
|
|
),
|
|
doc_type=str(raw.get("doc_type", "")),
|
|
tags=tuple(str(t) for t in (raw.get("tags") or []) if isinstance(t, str)),
|
|
content_ref=self._deploy_root / content_ref_raw,
|
|
chunk_ref=self._deploy_root / chunk_ref_raw,
|
|
artifact_ref=self._deploy_root / artifact_ref_raw,
|
|
review_status=str(raw.get("review_status", "")),
|
|
excluded_from_retrieval=bool(raw.get("excluded_from_retrieval", False)),
|
|
)
|
|
self._by_doc_id[doc.document_id] = doc
|
|
self._by_source_path[(doc.source_repo, doc.source_path)] = doc
|
|
self._by_repo.setdefault(doc.source_repo, []).append(doc)
|
|
loaded += 1
|
|
return loaded
|
|
|
|
def by_id(self, document_id: str) -> PublishedDocument | None:
|
|
return self._by_doc_id.get(document_id)
|
|
|
|
def by_source_path(self, repo: str, path: str) -> PublishedDocument | None:
|
|
return self._by_source_path.get((repo, path))
|
|
|
|
def documents_for_repo(self, repo: str) -> list[PublishedDocument]:
|
|
return list(self._by_repo.get(repo, []))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pack composition
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class PackSection:
|
|
section_id: str
|
|
title: str
|
|
tier: str
|
|
source_repo: str
|
|
document_id: str
|
|
source_path: str
|
|
source_ref: str | None
|
|
chunk_id: str | None
|
|
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:
|
|
return max(1, int(len(self.text) / CHARS_PER_TOKEN))
|
|
|
|
|
|
def _read_content(doc: PublishedDocument) -> str:
|
|
try:
|
|
return doc.content_ref.read_text(encoding="utf-8")
|
|
except OSError:
|
|
return ""
|
|
|
|
|
|
def _build_module_sections(
|
|
store: PublishRecordStore,
|
|
inputs: PackInputs,
|
|
module_map: ModuleMap,
|
|
) -> list[PackSection]:
|
|
if not inputs.changed_files:
|
|
return []
|
|
seen: set[tuple[str, str]] = set()
|
|
sections: list[PackSection] = []
|
|
for changed in inputs.changed_files:
|
|
modules = module_map.resolve(changed)
|
|
for module in modules:
|
|
for source_path in [module.spec, *module.adrs]:
|
|
if not source_path:
|
|
continue
|
|
key = (inputs.repo, source_path)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
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}",
|
|
title=doc.title or source_path,
|
|
tier=doc.tier if doc.tier != "unknown" else "ssot",
|
|
source_repo=doc.source_repo,
|
|
document_id=doc.document_id,
|
|
source_path=doc.source_path,
|
|
source_ref=doc.source_ref,
|
|
chunk_id=None,
|
|
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
|
|
|
|
|
|
def _build_dify_sections(
|
|
store: PublishRecordStore,
|
|
inputs: PackInputs,
|
|
dify: DifyClient,
|
|
dataset_ids: list[str],
|
|
query: str,
|
|
) -> list[PackSection]:
|
|
sections: list[PackSection] = []
|
|
seen: set[str] = set()
|
|
for dataset_id in dataset_ids:
|
|
# Try semantic first; economy datasets (no embedding model
|
|
# configured) fail with "Default model not found for text-embedding"
|
|
# — fall back to keyword_search for those.
|
|
hits: list[DifyRetrievalHit] = []
|
|
for method in ("semantic_search", "keyword_search"):
|
|
try:
|
|
hits = dify.retrieve(
|
|
dataset_id=dataset_id,
|
|
query=query,
|
|
top_k=inputs.top_k,
|
|
search_method=method,
|
|
)
|
|
break
|
|
except DifyClientError as exc:
|
|
msg = str(exc)
|
|
if method == "semantic_search" and (
|
|
"text-embedding" in msg or "embedding_model" in msg
|
|
):
|
|
continue # try keyword
|
|
print(
|
|
f"[context_pack] Dify retrieve skipped ({dataset_id} via {method}): {exc}",
|
|
file=sys.stderr,
|
|
)
|
|
hits = []
|
|
break
|
|
for hit in hits:
|
|
doc = _resolve_hit_to_local_doc(store, hit)
|
|
if doc is None:
|
|
continue
|
|
if not doc.is_retrievable():
|
|
continue
|
|
if doc.source_repo != inputs.repo:
|
|
continue
|
|
if hit.chunk_id in seen:
|
|
continue
|
|
seen.add(hit.chunk_id)
|
|
sections.append(
|
|
PackSection(
|
|
section_id=f"dify:{hit.chunk_id[:12]}",
|
|
title=doc.title or doc.source_path,
|
|
tier=doc.tier,
|
|
source_repo=doc.source_repo,
|
|
document_id=doc.document_id,
|
|
source_path=doc.source_path,
|
|
source_ref=doc.source_ref,
|
|
chunk_id=hit.chunk_id,
|
|
text=hit.content,
|
|
origin="dify",
|
|
score=hit.score,
|
|
)
|
|
)
|
|
return sections
|
|
|
|
|
|
def _resolve_hit_to_local_doc(
|
|
store: PublishRecordStore, hit: DifyRetrievalHit
|
|
) -> PublishedDocument | None:
|
|
"""Map a Dify hit back to a local publish record.
|
|
|
|
Dify document names equal the published document_id (per
|
|
ingest_artifact.py: document_name=document_id).
|
|
"""
|
|
return store.by_id(hit.document_name)
|
|
|
|
|
|
def _build_keyword_sections(
|
|
store: PublishRecordStore,
|
|
inputs: PackInputs,
|
|
query: str,
|
|
) -> list[PackSection]:
|
|
"""Fallback when Dify unavailable: naive token-overlap scoring on
|
|
locally-stored content."""
|
|
|
|
if not query.strip():
|
|
return []
|
|
terms = [t for t in re.split(r"\W+", query.lower()) if len(t) >= 2]
|
|
if not terms:
|
|
return []
|
|
candidates: list[tuple[float, PublishedDocument, str]] = []
|
|
for doc in store.documents_for_repo(inputs.repo):
|
|
if not doc.is_retrievable():
|
|
continue
|
|
text = _read_content(doc)
|
|
if not text:
|
|
continue
|
|
lower = text.lower()
|
|
hit_count = sum(lower.count(term) for term in terms)
|
|
if hit_count == 0 and not any(term in doc.title.lower() for term in terms):
|
|
continue
|
|
title_bonus = 5.0 if any(term in doc.title.lower() for term in terms) else 0.0
|
|
score = float(hit_count) + title_bonus
|
|
candidates.append((score, doc, text))
|
|
candidates.sort(key=lambda row: row[0], reverse=True)
|
|
|
|
sections: list[PackSection] = []
|
|
for score, doc, text in candidates[: inputs.top_k]:
|
|
sections.append(
|
|
PackSection(
|
|
section_id=f"keyword:{doc.document_id[:24]}",
|
|
title=doc.title or doc.source_path,
|
|
tier=doc.tier,
|
|
source_repo=doc.source_repo,
|
|
document_id=doc.document_id,
|
|
source_path=doc.source_path,
|
|
source_ref=doc.source_ref,
|
|
chunk_id=None,
|
|
text=text,
|
|
origin="task-keyword",
|
|
score=score,
|
|
)
|
|
)
|
|
return sections
|
|
|
|
|
|
def _dedupe(sections: list[PackSection]) -> list[PackSection]:
|
|
seen: set[tuple[str, str | None]] = set()
|
|
out: list[PackSection] = []
|
|
for s in sections:
|
|
key = (s.document_id, s.chunk_id)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
out.append(s)
|
|
return out
|
|
|
|
|
|
def _rank(sections: list[PackSection]) -> list[PackSection]:
|
|
return sorted(
|
|
sections,
|
|
key=lambda s: (
|
|
TIER_ORDER.get(s.tier, TIER_ORDER["unknown"]),
|
|
0 if s.origin == "module-map" else 1,
|
|
-s.score,
|
|
),
|
|
)
|
|
|
|
|
|
def _cap_by_budget(sections: list[PackSection], budget: int) -> list[PackSection]:
|
|
out: list[PackSection] = []
|
|
used = 0
|
|
for s in sections:
|
|
cost = s.token_estimate
|
|
if used + cost > budget:
|
|
# Allow partial inclusion of the first overflow if pack is empty,
|
|
# so we never produce a completely empty pack just because the
|
|
# first doc is huge.
|
|
if not out:
|
|
out.append(_truncate(s, budget))
|
|
break
|
|
out.append(s)
|
|
used += cost
|
|
return out
|
|
|
|
|
|
def _truncate(section: PackSection, max_tokens: int) -> PackSection:
|
|
max_chars = int(max_tokens * CHARS_PER_TOKEN)
|
|
if len(section.text) <= max_chars:
|
|
return section
|
|
truncated = section.text[: max_chars - 60].rstrip()
|
|
truncated += "\n\n…(truncated to fit token budget; cite chunk_id for full text)"
|
|
return PackSection(
|
|
section_id=section.section_id,
|
|
title=section.title,
|
|
tier=section.tier,
|
|
source_repo=section.source_repo,
|
|
document_id=section.document_id,
|
|
source_path=section.source_path,
|
|
source_ref=section.source_ref,
|
|
chunk_id=section.chunk_id,
|
|
text=truncated,
|
|
origin=section.origin,
|
|
score=section.score,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Output writers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _format_markdown(inputs: PackInputs, sections: list[PackSection], header: dict) -> str:
|
|
lines: list[str] = []
|
|
lines.append("# Knowledge Context Pack")
|
|
lines.append("")
|
|
lines.append(f"- repo: {inputs.repo}")
|
|
lines.append(f"- mode: {inputs.mode}")
|
|
lines.append(f"- generated_at: {header['generated_at']}")
|
|
lines.append(f"- query: {header['query'] or '(none)'}")
|
|
lines.append(f"- changed_files: {len(inputs.changed_files)}")
|
|
lines.append(f"- sections: {len(sections)}")
|
|
lines.append(f"- token_budget: {inputs.token_budget}")
|
|
lines.append(f"- token_estimate: {header['token_estimate']}")
|
|
lines.append(f"- dify_query_at: {header['dify_query_at'] or '(n/a)'}")
|
|
lines.append(f"- spec_set_hash: {header['spec_set_hash']}")
|
|
lines.append("")
|
|
lines.append("Citation grammar: cite each claim with one of")
|
|
lines.append(" `[kg:repo:document_id:chunk_id]` (chunk-level, strongest),")
|
|
lines.append(" `[kg:repo:source_path]` (document-level),")
|
|
lines.append(" `source_path:Lx-Ly` (source file & line range).")
|
|
lines.append("")
|
|
if not sections:
|
|
lines.append("> No retrievable context for this query / changed-files.")
|
|
lines.append("> Read the working tree directly and proceed with caution.")
|
|
return "\n".join(lines) + "\n"
|
|
|
|
for index, s in enumerate(sections, start=1):
|
|
lines.append(f"## {index}. {s.title}")
|
|
lines.append("")
|
|
lines.append(f"- tier: `{s.tier}`")
|
|
lines.append(f"- source_path: `{s.source_path}`")
|
|
if s.source_ref:
|
|
lines.append(f"- source_ref: `{s.source_ref}`")
|
|
if s.chunk_id:
|
|
lines.append(f"- chunk_id: `{s.chunk_id}`")
|
|
lines.append(f"- document_id: `{s.document_id}`")
|
|
lines.append(f"- origin: {s.origin} score: {s.score:.2f}")
|
|
cite = f"[kg:{s.source_repo}:{s.document_id}:{s.chunk_id}]" if s.chunk_id else f"[kg:{s.source_repo}:{s.source_path}]"
|
|
lines.append(f"- cite as: `{cite}`")
|
|
lines.append("")
|
|
lines.append(s.text.rstrip())
|
|
lines.append("")
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def _format_manifest(
|
|
inputs: PackInputs, sections: list[PackSection], header: dict
|
|
) -> dict:
|
|
return {
|
|
"schema_version": "context-pack-1",
|
|
"repo": inputs.repo,
|
|
"mode": inputs.mode,
|
|
"generated_at": header["generated_at"],
|
|
"query": header["query"],
|
|
"changed_files": list(inputs.changed_files),
|
|
"token_budget": inputs.token_budget,
|
|
"token_estimate": header["token_estimate"],
|
|
"dify_query_at": header["dify_query_at"],
|
|
"spec_set_hash": header["spec_set_hash"],
|
|
"ssot_hits": sum(1 for s in sections if s.tier == "ssot"),
|
|
"tier_counts": {
|
|
tier: sum(1 for s in sections if s.tier == tier)
|
|
for tier in ("ssot", "proposed", "reference", "legacy", "unknown")
|
|
},
|
|
"sections": [
|
|
{
|
|
"section_id": s.section_id,
|
|
"title": s.title,
|
|
"tier": s.tier,
|
|
"source_repo": s.source_repo,
|
|
"document_id": s.document_id,
|
|
"source_path": s.source_path,
|
|
"source_ref": s.source_ref,
|
|
"chunk_id": s.chunk_id,
|
|
"origin": s.origin,
|
|
"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
|
|
],
|
|
}
|
|
|
|
|
|
def _spec_set_hash(sections: list[PackSection]) -> str:
|
|
"""Hash the concrete source set included in the pack."""
|
|
|
|
payload = [
|
|
{
|
|
"document_id": s.document_id,
|
|
"source_path": s.source_path,
|
|
"source_ref": s.source_ref,
|
|
"tier": s.tier,
|
|
"text_sha256": sha256_text(s.text),
|
|
}
|
|
for s in sections
|
|
]
|
|
return sha256_text(
|
|
json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _qualify_repo(repo: str) -> str:
|
|
"""Convert `service-name` to the `owner__repo` directory qualifier.
|
|
|
|
Accepts already-qualified `owner__repo` form too. Defaults owner to
|
|
`lingjingwanxiang` when only the service name is supplied.
|
|
"""
|
|
|
|
if "__" in repo:
|
|
return repo
|
|
return f"lingjingwanxiang__{repo}"
|
|
|
|
|
|
def _strip_repo_owner(repo_qualifier: str) -> str:
|
|
"""`lingjingwanxiang__ljwx-platform` -> `lingjingwanxiang/ljwx-platform`."""
|
|
if "__" not in repo_qualifier:
|
|
return repo_qualifier
|
|
owner, name = repo_qualifier.split("__", 1)
|
|
return f"{owner}/{name}"
|
|
|
|
|
|
def _resolve_dify(
|
|
deploy_root: Path,
|
|
) -> tuple[DifyClient | None, list[str]]:
|
|
base_url = os.environ.get("KNOWLEDGE_DIFY_BASE_URL", "").strip()
|
|
api_key = os.environ.get("KNOWLEDGE_DIFY_API_KEY", "").strip()
|
|
if not base_url or not api_key:
|
|
return None, []
|
|
try:
|
|
client = DifyClient(base_url=base_url, api_key=api_key)
|
|
except DifyClientError as exc:
|
|
print(f"[context_pack] Dify client init failed: {exc}", file=sys.stderr)
|
|
return None, []
|
|
# Collect dataset ids from publish-targets via stored records: any
|
|
# `targets[].dataset_id` seen in publish-records is a candidate. For
|
|
# the MVP we look up by name from publish-targets.yaml.
|
|
targets_path = deploy_root / "platform" / "knowledge" / "publish-targets.yaml"
|
|
if not targets_path.exists():
|
|
return client, []
|
|
raw = load_yaml_mapping(targets_path)
|
|
raw_targets = raw.get("targets", {})
|
|
names: list[str] = []
|
|
if isinstance(raw_targets, dict):
|
|
for cfg in raw_targets.values():
|
|
if isinstance(cfg, dict) and cfg.get("type") == "dify_dataset":
|
|
ds_name = cfg.get("dataset_name")
|
|
if isinstance(ds_name, str) and ds_name:
|
|
names.append(ds_name)
|
|
dataset_ids: list[str] = []
|
|
try:
|
|
live = {ds.name: ds.dataset_id for ds in client.list_datasets()}
|
|
except DifyClientError as exc:
|
|
print(f"[context_pack] Dify list_datasets skipped: {exc}", file=sys.stderr)
|
|
return client, []
|
|
for name in names:
|
|
if name in live:
|
|
dataset_ids.append(live[name])
|
|
return client, dataset_ids
|
|
|
|
|
|
def _read_changed_files(path: Path | None) -> tuple[str, ...]:
|
|
if path is None:
|
|
return ()
|
|
if not path.exists():
|
|
raise KnowledgeContractError(f"--changed-files not found: {path}")
|
|
lines = []
|
|
for raw in path.read_text(encoding="utf-8").splitlines():
|
|
raw = raw.strip()
|
|
if not raw:
|
|
continue
|
|
# Accept `git diff --name-status` output (M\tpath) or plain paths.
|
|
parts = raw.split("\t")
|
|
if len(parts) >= 2:
|
|
lines.append(parts[-1])
|
|
else:
|
|
lines.append(parts[0])
|
|
return tuple(lines)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Build a knowledge context pack")
|
|
parser.add_argument("--repo", required=True, help="service name or owner__repo")
|
|
parser.add_argument(
|
|
"--mode", choices=["implement", "review"], default="implement"
|
|
)
|
|
parser.add_argument("--task-text", default="")
|
|
parser.add_argument("--changed-files", type=Path, default=None)
|
|
parser.add_argument(
|
|
"--repo-root",
|
|
type=Path,
|
|
default=None,
|
|
help="business repo root (for knowledge.module-map.yaml); defaults to "
|
|
"../<repo-name>",
|
|
)
|
|
parser.add_argument(
|
|
"--deploy-root",
|
|
type=Path,
|
|
default=Path.cwd(),
|
|
help="ljwx-deploy checkout root",
|
|
)
|
|
parser.add_argument(
|
|
"--output",
|
|
type=Path,
|
|
default=Path(".knowledge-pack"),
|
|
help="output directory for pack.md + pack.json",
|
|
)
|
|
parser.add_argument("--env", default="dev")
|
|
parser.add_argument("--token-budget", type=int, default=DEFAULT_TOKEN_BUDGET)
|
|
parser.add_argument("--top-k", type=int, default=DEFAULT_TOP_K)
|
|
parser.add_argument(
|
|
"--skip-dify",
|
|
action="store_true",
|
|
help="ignore Dify even when env vars are present",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
repo_qualifier = _qualify_repo(args.repo)
|
|
repo_owner_slash = _strip_repo_owner(repo_qualifier)
|
|
service_name = repo_qualifier.split("__", 1)[1]
|
|
|
|
repo_root = args.repo_root
|
|
if repo_root is None:
|
|
candidate = Path.cwd().parent / service_name
|
|
repo_root = candidate if candidate.exists() else Path.cwd()
|
|
|
|
inputs = PackInputs(
|
|
repo=repo_owner_slash,
|
|
mode=args.mode,
|
|
task_text=args.task_text or None,
|
|
changed_files=_read_changed_files(args.changed_files),
|
|
repo_root=repo_root,
|
|
deploy_root=args.deploy_root,
|
|
output_dir=args.output,
|
|
env=args.env,
|
|
token_budget=args.token_budget,
|
|
top_k=args.top_k,
|
|
)
|
|
|
|
store = PublishRecordStore(inputs.deploy_root, inputs.env)
|
|
loaded = store.load_for_repo(repo_qualifier)
|
|
if loaded == 0:
|
|
print(
|
|
f"[context_pack] no publish records under {repo_qualifier}; pack will be empty unless Dify hits",
|
|
file=sys.stderr,
|
|
)
|
|
|
|
module_map = load_module_map(inputs.repo_root)
|
|
if module_map is None:
|
|
print(
|
|
f"[context_pack] {inputs.repo_root}/knowledge.module-map.yaml not found; module resolution skipped",
|
|
file=sys.stderr,
|
|
)
|
|
|
|
sections: list[PackSection] = []
|
|
if module_map is not None:
|
|
sections.extend(_build_module_sections(store, inputs, module_map))
|
|
|
|
# Dify's /retrieve caps the query at 250 chars. Prefer task_text;
|
|
# only append changed-files if there's room. Trailing ellipsis
|
|
# signals truncation to ops in Dify access logs.
|
|
DIFY_QUERY_MAX = 240
|
|
parts: list[str] = []
|
|
if inputs.task_text:
|
|
parts.append(inputs.task_text)
|
|
for path in inputs.changed_files:
|
|
# Just the basename — that's the highest-signal token for
|
|
# retrieval; full path eats the budget.
|
|
parts.append(Path(path).name)
|
|
query = " ".join(parts).strip()
|
|
if len(query) > DIFY_QUERY_MAX:
|
|
query = query[: DIFY_QUERY_MAX - 1].rstrip() + "…"
|
|
|
|
dify_query_at: str | None = None
|
|
if not args.skip_dify and query:
|
|
client, dataset_ids = _resolve_dify(inputs.deploy_root)
|
|
if client is not None and dataset_ids:
|
|
dify_sections = _build_dify_sections(
|
|
store=store,
|
|
inputs=inputs,
|
|
dify=client,
|
|
dataset_ids=dataset_ids,
|
|
query=query,
|
|
)
|
|
sections.extend(dify_sections)
|
|
dify_query_at = utc_now()
|
|
|
|
if query and (
|
|
not any(s.origin == "dify" for s in sections)
|
|
or not any(s.tier == "ssot" for s in sections)
|
|
):
|
|
# Either Dify off, Dify returned nothing for this repo, or Dify only
|
|
# returned reference material. Add local keyword candidates so
|
|
# review/implement packs still have a chance to include SSOT sources.
|
|
sections.extend(_build_keyword_sections(store, inputs, query))
|
|
|
|
sections = _dedupe(sections)
|
|
sections = _rank(sections)
|
|
sections = _cap_by_budget(sections, inputs.token_budget)
|
|
|
|
generated_at = utc_now()
|
|
token_estimate = sum(s.token_estimate for s in sections)
|
|
header = {
|
|
"generated_at": generated_at,
|
|
"query": query or None,
|
|
"token_estimate": token_estimate,
|
|
"dify_query_at": dify_query_at,
|
|
"spec_set_hash": _spec_set_hash(sections),
|
|
}
|
|
|
|
inputs.output_dir.mkdir(parents=True, exist_ok=True)
|
|
md_path = inputs.output_dir / "context-pack.md"
|
|
json_path = inputs.output_dir / "context-pack.json"
|
|
md_path.write_text(_format_markdown(inputs, sections, header), encoding="utf-8")
|
|
write_json(json_path, _format_manifest(inputs, sections, header))
|
|
|
|
summary = {
|
|
"sections": len(sections),
|
|
"ssot_hits": sum(1 for s in sections if s.tier == "ssot"),
|
|
"token_estimate": token_estimate,
|
|
"dify_used": dify_query_at is not None,
|
|
"output_md": str(md_path),
|
|
"output_json": str(json_path),
|
|
}
|
|
print(json.dumps(summary, ensure_ascii=False))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|