Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions doc/architecture/SPEC/kb_service.md
Original file line number Diff line number Diff line change
Expand Up @@ -311,3 +311,14 @@ x 必须为正。因阈值停下的调用(`stop_reason="max_budget"`)→ `Mo
不传上限的现有调用方行为不变:该参数根本不会发给 transport。
`runtime.trace_recorder` 把阈值与花费持久化到 `model_call` 记录的 `result.max_budget_usd` / `result.cost_usd`
(未设或未知为 null),包括 transport 抛错的失败记录。

## 2026-09-30 `kb init` 的 adapter 配置
`knowledge_lifecycle.init`(可选子映射,严格校验,未知键抛 `LifecycleConfigError`)解析为 `InitConfig`,挂在
`RepoLifecycle.init`(缺省为 `None`):`seeds`(知识库相对路径,只能是 `general/...` 或 `repos/<其他仓库>/...`,
不得指向本仓库、不得为绝对路径或含空段/`.`/`..`,不得重复)、`doc_globs`(缺省 `DEFAULT_DOC_GLOBS` =
`README*`、`docs/**/*.md`(与 `profiles/establish.build_doc_corpus` 一致)加 `CONTRIBUTING.md`、`AGENTS.md`)、
`source_roots`(缺省取同一 manifest 的 `ut_coverage.source_roots`)、`exclude`、`module_depth`(≥1,缺省 2)、
`min_module_loc`(≥0,缺省 300)、`pr_window: {count, max_age_days}`(>0,缺省 200 / 180)、`coverage_target`
((0, 1],缺省 0.85)、`budget_usd`(>0,缺省 30)、`judge_call_usd`(≥0,缺省 0.50)。`validate_seeds(init,
knowledge_root)` 是运行时检查:种子必须是知识树中已存在的文件或目录,解析后不得逃出知识根目录,逐条返回问题。
解析只做静态校验,不读知识树。测试:`test_kb_init_config.py`。
118 changes: 117 additions & 1 deletion src/infermatrix_copilot/kb_service/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@

from __future__ import annotations

import math
import re
from dataclasses import dataclass, field
from pathlib import Path
Expand All @@ -23,8 +24,15 @@
_KEYS = {
"enabled", "mode", "knowledge_dir", "intake", "release", "sweep",
"upstream_visibility", "circuit_breaker", "protected_rules",
"calibration_set", "model_dir",
"calibration_set", "model_dir", "init",
}
_INIT_KEYS = {
"seeds", "doc_globs", "source_roots", "exclude", "module_depth", "min_module_loc",
"pr_window", "coverage_target", "budget_usd", "judge_call_usd",
}
# README* + docs/**/*.md is what profiles/establish.build_doc_corpus reads;
# the contribution and agent guides carry most cross-doc invariants.
DEFAULT_DOC_GLOBS = ("README*", "docs/**/*.md", "CONTRIBUTING.md", "AGENTS.md")


class LifecycleConfigError(ValueError):
Expand All @@ -46,6 +54,22 @@ class ReleaseConfig:
auditor: str = "" # plugin path relative to the adapter directory


@dataclass(frozen=True)
class InitConfig:
"""How ``kb init`` bootstraps the repository's knowledge (design kb-init §3)."""
seeds: tuple[str, ...] = () # knowledge-relative: general/... or repos/<other>/...
doc_globs: tuple[str, ...] = DEFAULT_DOC_GLOBS
source_roots: tuple[str, ...] = ()
exclude: tuple[str, ...] = ()
module_depth: int = 2
min_module_loc: int = 300
pr_window_count: int = 200
pr_window_max_age_days: int = 180
coverage_target: float = 0.85
budget_usd: float = 30.0
judge_call_usd: float = 0.50


@dataclass(frozen=True)
class RepoLifecycle:
repo: str # knowledge repo name, or "general"
Expand All @@ -63,6 +87,7 @@ class RepoLifecycle:
calibration_set: str = "" # directory relative to the adapter (ships in the wheel)
model_dir: str = ""
adapter_dir: Path | None = None
init: InitConfig | None = None

@property
def auto_merge(self) -> bool:
Expand Down Expand Up @@ -92,6 +117,96 @@ def _mapping(value: object, where: str, allowed: set[str]) -> dict:
return value


def _strings(value: object, where: str) -> tuple[str, ...]:
if value is None:
return ()
if not isinstance(value, list) or any(not isinstance(v, str) or not v.strip() for v in value):
raise LifecycleConfigError(f"{where} must be a list of non-empty strings")
return tuple(v.strip() for v in value)


def _int(value: object, where: str, *, minimum: int) -> int:
if isinstance(value, bool) or not isinstance(value, int) or value < minimum:
raise LifecycleConfigError(f"{where} must be an integer >= {minimum}")
return value


def _number(value: object, where: str) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
raise LifecycleConfigError(f"{where} must be a finite number")
return float(value)


def _seed_problem(seed: str, repo: str) -> str:
parts = seed.split("/")
if seed.startswith("/") or "\\" in seed or any(p in ("", ".", "..") for p in parts):
return "must be a relative knowledge path without empty, . or .. segments"
if parts[0] == GENERAL:
return ""
if parts[0] == "repos" and len(parts) >= 2 and _NAME.fullmatch(parts[1]):
return "names the repository being initialised" if parts[1] == repo else ""
return "must start with general/ or repos/<other repo>/"


def parse_init(section: object, where: str, *, repo: str, manifest: dict) -> InitConfig | None:
"""The ``init`` sub-mapping, strictly validated; None when absent."""
if section is None:
return None
data = _mapping(section, where, _INIT_KEYS)
seeds = _strings(data.get("seeds"), f"{where}.seeds")
for seed in seeds:
problem = _seed_problem(seed, repo)
if problem:
raise LifecycleConfigError(f"{where}.seeds: {seed!r} {problem}")
if len(set(seeds)) != len(seeds):
raise LifecycleConfigError(f"{where}.seeds has duplicates")
if "source_roots" in data:
source_roots = _strings(data["source_roots"], f"{where}.source_roots")
else:
ut = manifest.get("ut_coverage") if isinstance(manifest.get("ut_coverage"), dict) else {}
roots = ut.get("source_roots") if isinstance(ut.get("source_roots"), list) else []
source_roots = tuple(str(r).strip() for r in roots if isinstance(r, str) and r.strip())
window = _mapping(data.get("pr_window"), f"{where}.pr_window", {"count", "max_age_days"})
coverage = _number(data.get("coverage_target", 0.85), f"{where}.coverage_target")
if not 0 < coverage <= 1:
raise LifecycleConfigError(f"{where}.coverage_target must be in (0, 1]")
budget = _number(data.get("budget_usd", 30.0), f"{where}.budget_usd")
if budget <= 0:
raise LifecycleConfigError(f"{where}.budget_usd must be > 0")
judge_call = _number(data.get("judge_call_usd", 0.50), f"{where}.judge_call_usd")
if judge_call < 0:
raise LifecycleConfigError(f"{where}.judge_call_usd must be >= 0")
return InitConfig(
seeds=seeds,
doc_globs=(_strings(data["doc_globs"], f"{where}.doc_globs") if "doc_globs" in data
else DEFAULT_DOC_GLOBS),
source_roots=source_roots,
exclude=_strings(data.get("exclude"), f"{where}.exclude"),
module_depth=_int(data.get("module_depth", 2), f"{where}.module_depth", minimum=1),
min_module_loc=_int(data.get("min_module_loc", 300), f"{where}.min_module_loc", minimum=0),
pr_window_count=_int(window.get("count", 200), f"{where}.pr_window.count", minimum=1),
pr_window_max_age_days=_int(window.get("max_age_days", 180),
f"{where}.pr_window.max_age_days", minimum=1),
coverage_target=coverage,
budget_usd=budget,
judge_call_usd=judge_call,
)


def validate_seeds(init: InitConfig, knowledge_root: Path) -> list[str]:
"""Run-time check: every seed names an existing file or directory under
the knowledge root. Returns one problem per missing seed."""
root = Path(knowledge_root).resolve()
problems = []
for seed in init.seeds:
target = (root / seed).resolve()
if not target.is_relative_to(root):
Comment on lines +202 to +203

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Reject symlink aliases that resolve into the initialized repo

When a valid-looking seed such as general/alias is a symlink to knowledge_root/repos/<repo>, _seed_problem approves the textual path and this runtime check approves the resolved target merely because it remains under the knowledge root. This bypasses the documented self-seed prohibition and can feed the repository's own existing knowledge back into initialization; runtime validation should also reject resolved paths under the repository being initialized, or reject such symlink aliases.

Useful? React with 👍 / 👎.

problems.append(f"seed {seed} escapes the knowledge root")
elif not target.exists():
problems.append(f"seed {seed} does not exist in the knowledge tree")
return problems


def parse_lifecycle(adapter: RepoAdapter) -> RepoLifecycle | None:
"""The adapter's lifecycle, or None when it has no knowledge_lifecycle."""
section = adapter.manifest.get("knowledge_lifecycle")
Expand Down Expand Up @@ -171,6 +286,7 @@ def parse_lifecycle(adapter: RepoAdapter) -> RepoLifecycle | None:
calibration_set=str(data.get("calibration_set") or ""),
model_dir=str(data.get("model_dir") or ""),
adapter_dir=Path(adapter.root),
init=parse_init(data.get("init"), f"{where}.init", repo=repo, manifest=adapter.manifest),
)


Expand Down
142 changes: 142 additions & 0 deletions test/test_kb_init_config.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,142 @@
"""kb init: the adapter's knowledge_lifecycle.init block (design kb-init §3)."""

from __future__ import annotations

from pathlib import Path

import pytest
import yaml

from infermatrix_copilot.adapters.base import RepoAdapter
from infermatrix_copilot.kb_service.config import (
DEFAULT_DOC_GLOBS, InitConfig, LifecycleConfigError, parse_lifecycle, validate_seeds,
)


def _adapter(tmp_path: Path, init: object = None, *, extra: dict | None = None,
with_init: bool = True) -> RepoAdapter:
root = tmp_path / "demo_adapter"
root.mkdir(exist_ok=True)
section: dict = {"enabled": False, "mode": "shadow"}
if with_init:
section["init"] = init
manifest = {
"name": "demo_adapter",
"repo": {"full_name": "org/demo", "path": "/x"},
"knowledge": {"repo_subdir": "repos/demo"},
"knowledge_lifecycle": section,
**(extra or {}),
}
(root / "manifest.yaml").write_text(yaml.safe_dump(manifest), encoding="utf-8")
return RepoAdapter(name="demo_adapter", root=root, manifest=manifest)


def test_absent_init_is_none(tmp_path):
assert parse_lifecycle(_adapter(tmp_path, with_init=False)).init is None


def test_empty_init_gets_defaults(tmp_path):
init = parse_lifecycle(_adapter(tmp_path, {})).init
assert init == InitConfig()
assert init.doc_globs == DEFAULT_DOC_GLOBS
assert (init.module_depth, init.min_module_loc) == (2, 300)
assert (init.pr_window_count, init.pr_window_max_age_days) == (200, 180)
assert (init.coverage_target, init.budget_usd, init.judge_call_usd) == (0.85, 30.0, 0.50)
assert init.seeds == init.source_roots == init.exclude == ()


def test_full_init_parses(tmp_path):
init = parse_lifecycle(_adapter(tmp_path, {
"seeds": ["general/review/x.md", "repos/other/ci"],
"doc_globs": ["README.md"],
"source_roots": ["pkg/"],
"exclude": ["*/tests/*"],
"module_depth": 3,
"min_module_loc": 0,
"pr_window": {"count": 50, "max_age_days": 30},
"coverage_target": 1,
"budget_usd": 5,
"judge_call_usd": 0,
})).init
assert init.seeds == ("general/review/x.md", "repos/other/ci")
assert init.doc_globs == ("README.md",)
assert init.source_roots == ("pkg/",)
assert init.exclude == ("*/tests/*",)
assert (init.module_depth, init.min_module_loc) == (3, 0)
assert (init.pr_window_count, init.pr_window_max_age_days) == (50, 30)
assert (init.coverage_target, init.budget_usd, init.judge_call_usd) == (1.0, 5.0, 0.0)


def test_source_roots_default_to_ut_coverage(tmp_path):
adapter = _adapter(tmp_path, {}, extra={"ut_coverage": {"source_roots": ["pkg/", "csrc/"]}})
assert parse_lifecycle(adapter).init.source_roots == ("pkg/", "csrc/")
adapter = _adapter(tmp_path, {"source_roots": ["only/"]},
extra={"ut_coverage": {"source_roots": ["pkg/"]}})
assert parse_lifecycle(adapter).init.source_roots == ("only/",)


@pytest.mark.parametrize("init, message", [
("nope", "must be a mapping"),
({"surprise": 1}, "unknown keys"),
({"pr_window": {"count": 1, "days": 2}}, "unknown keys"),
({"seeds": "general/x.md"}, "list of non-empty strings"),
({"seeds": [""]}, "list of non-empty strings"),
({"seeds": ["general/x.md", "general/x.md"]}, "duplicates"),
({"seeds": ["repos/demo/rules.md"]}, "names the repository being initialised"),
({"seeds": ["/abs/general/x.md"]}, "relative knowledge path"),
({"seeds": ["general/../repos/demo/x.md"]}, "relative knowledge path"),
({"seeds": ["general//x.md"]}, "relative knowledge path"),
({"seeds": ["skills/x.md"]}, "general/ or repos/"),
({"seeds": ["repos"]}, "general/ or repos/"),
({"seeds": ["repos/Bad Name/x.md"]}, "general/ or repos/"),
({"doc_globs": [1]}, "list of non-empty strings"),
({"exclude": "x"}, "list of non-empty strings"),
({"module_depth": 0}, "module_depth must be an integer >= 1"),
({"module_depth": True}, "module_depth must be an integer"),
({"min_module_loc": -1}, "min_module_loc must be an integer >= 0"),
({"min_module_loc": 1.5}, "min_module_loc must be an integer"),
({"pr_window": {"count": 0}}, "pr_window.count"),
({"pr_window": {"max_age_days": 0}}, "pr_window.max_age_days"),
({"coverage_target": 0}, "coverage_target must be in"),
({"coverage_target": 1.01}, "coverage_target must be in"),
({"coverage_target": "high"}, "coverage_target must be a finite number"),
({"budget_usd": 0}, "budget_usd must be > 0"),
({"budget_usd": True}, "budget_usd must be a finite number"),
({"judge_call_usd": -0.1}, "judge_call_usd must be >= 0"),
({"budget_usd": float("nan")}, "budget_usd must be a finite number"),
({"budget_usd": float("inf")}, "budget_usd must be a finite number"),
({"judge_call_usd": float("nan")}, "judge_call_usd must be a finite number"),
({"judge_call_usd": float("inf")}, "judge_call_usd must be a finite number"),
({"coverage_target": float("nan")}, "coverage_target must be a finite number"),
])
def test_invalid_init_is_rejected(tmp_path, init, message):
with pytest.raises(LifecycleConfigError, match=message):
parse_lifecycle(_adapter(tmp_path, init))


def test_non_finite_yaml_values_are_rejected(tmp_path):
adapter = _adapter(tmp_path, {})
adapter.manifest["knowledge_lifecycle"]["init"] = yaml.safe_load("budget_usd: .inf")
with pytest.raises(LifecycleConfigError, match="finite"):
parse_lifecycle(adapter)


def test_validate_seeds_reports_missing_paths(tmp_path):
root = tmp_path / "knowledge"
(root / "general" / "review").mkdir(parents=True)
(root / "general" / "review" / "x.md").write_text("x", encoding="utf-8")
(root / "repos" / "other" / "ci").mkdir(parents=True)
ok = InitConfig(seeds=("general/review/x.md", "repos/other/ci"))
assert validate_seeds(ok, root) == []
missing = InitConfig(seeds=("general/review/x.md", "repos/other/gone.md"))
assert validate_seeds(missing, root) == ["seed repos/other/gone.md does not exist in the knowledge tree"]


def test_validate_seeds_refuses_escape(tmp_path):
root = tmp_path / "knowledge"
root.mkdir()
(tmp_path / "outside.md").write_text("x", encoding="utf-8")
(root / "general").mkdir()
(root / "general" / "link.md").symlink_to(tmp_path / "outside.md")
assert validate_seeds(InitConfig(seeds=("general/link.md",)), root) == [
"seed general/link.md escapes the knowledge root"]
Loading