From eeaa3093fd2d0ac5edffb602db823a4b717f4ecf Mon Sep 17 00:00:00 2001 From: tzhouam Date: Mon, 28 Sep 2026 19:12:20 +0800 Subject: [PATCH] Add the knowledge service core: repo registry, ledger, signed outbox Foundation for the knowledge service that takes knowledge ownership from ReviewBot. There are no workflows or scheduler yet; those build on this. - kb_service.config parses each adapter's knowledge_lifecycle section into a repo registry (general/ is configured by the service). It refuses unknown keys, bad modes or triggers, a missing auditor plugin, a knowledge_dir that disagrees with the adapter, auto_merge without a calibration set, and any private upstream in auto_merge. knowledge_lifecycle joins the adapter HIGH_RISK_SECTIONS, so only a human can enable auto-merge. - kb_service.ledger: one SQLite file with every table keyed by repo, a heartbeat lease for single-instance operation, and per-repo and global generation numbers; bump_generation voids everything issued before it. - knowledge_service.signing: Ed25519 envelopes whose signature covers a purpose tag, so a verdict, outbox item, control record, hold list or ack signature never verifies for another purpose. Keys are 0600 and never overwritten. Vendorable (only cryptography, new `kb` extra). - kb_service.outbox: the service writes signed items, a control record and a hold list, and reads acks signed by the publisher's key. check_item() is what the GPU-box publisher runs before any write: signature, control record no older than 10 minutes, expiry, generation, pause, private upstream, and no enqueue or verdict unless the repo is in auto_merge; pause and close are always allowed. verify_holds() fails closed on a stale list. - `infermatrix-copilot kb keygen|status|pause|resume|control`. - vllm-omni gets knowledge_lifecycle in shadow mode with its release_audit.py auditor; afd-plugin is declared but disabled until its own calibration. - CI installs the kb extra. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/test.yml | 5 +- adapters/afd_plugin/manifest.yaml | 10 + adapters/vllm_omni/manifest.yaml | 23 + doc/architecture/SPEC/adapters/base.md | 3 + doc/architecture/SPEC/cli.md | 5 +- doc/architecture/SPEC/kb_service.md | 29 ++ doc/architecture/SPEC/knowledge_service.md | 5 + doc/features/knowledge-service.md | 41 ++ pyproject.toml | 2 + src/infermatrix_copilot/adapters/base.py | 2 +- src/infermatrix_copilot/cli/entry.py | 6 + .../kb_service/__init__.py | 1 + src/infermatrix_copilot/kb_service/cli.py | 145 +++++++ src/infermatrix_copilot/kb_service/config.py | 210 +++++++++ src/infermatrix_copilot/kb_service/ledger.py | 392 +++++++++++++++++ src/infermatrix_copilot/kb_service/outbox.py | 283 +++++++++++++ .../knowledge_service/signing.py | 117 ++++++ test/test_kb_service_core.py | 397 ++++++++++++++++++ 18 files changed, 1672 insertions(+), 4 deletions(-) create mode 100644 doc/architecture/SPEC/kb_service.md create mode 100644 doc/features/knowledge-service.md create mode 100644 src/infermatrix_copilot/kb_service/__init__.py create mode 100644 src/infermatrix_copilot/kb_service/cli.py create mode 100644 src/infermatrix_copilot/kb_service/config.py create mode 100644 src/infermatrix_copilot/kb_service/ledger.py create mode 100644 src/infermatrix_copilot/kb_service/outbox.py create mode 100644 src/infermatrix_copilot/knowledge_service/signing.py create mode 100644 test/test_kb_service_core.py diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 697daa2bb..1c65dc5c3 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -36,9 +36,10 @@ jobs: with: python-version: "3.12" - # pytest lives in the `dev` extra; `mcp` is needed for the MCP server tests. + # pytest lives in the `dev` extra; `mcp` is needed for the MCP server tests; + # `kb` (cryptography) for the knowledge service's signing. - name: Install - run: pip install -e ".[dev,mcp]" + run: pip install -e ".[dev,mcp,kb]" - name: Unit tests run: python -m pytest -q diff --git a/adapters/afd_plugin/manifest.yaml b/adapters/afd_plugin/manifest.yaml index 0ac7aea56..23245337b 100644 --- a/adapters/afd_plugin/manifest.yaml +++ b/adapters/afd_plugin/manifest.yaml @@ -224,3 +224,13 @@ push: default_remote: origin protected_branches: [main, master] allowed: false + +# Knowledge service: declared but disabled until afd-plugin has its own +# calibration set and has run through a shadow period (rollout stage P3b). +knowledge_lifecycle: + enabled: false + mode: shadow + knowledge_dir: repos/afd-plugin + release: + trigger: none + upstream_visibility: public diff --git a/adapters/vllm_omni/manifest.yaml b/adapters/vllm_omni/manifest.yaml index e8318f1b1..0d86e46e4 100644 --- a/adapters/vllm_omni/manifest.yaml +++ b/adapters/vllm_omni/manifest.yaml @@ -399,3 +399,26 @@ push: rebase_branch: dev/vllm-align # rebase-commit author identity (parent config SIGNOFF_NAME/EMAIL) signoff: {name: tzhouam, email: tzhouam@connect.ust.hk} + +# Knowledge service (doc/features/knowledge-service.md). Human-only section: +# switching `mode` to auto_merge is a reviewed code change, never an agent write. +knowledge_lifecycle: + enabled: true + mode: shadow # shadow | auto_merge + knowledge_dir: repos/vllm-omni + intake: + merged_prs: true + copilot_runs: true + human_prs: true + daily_model_budget_usd: 50 + release: + trigger: github_release + auditor: release_audit.py # adapters/vllm_omni/release_audit.py + sweep: + fallback_interval_days: 30 + upstream_visibility: public + circuit_breaker: + retire_ratio: 0.10 + max_files: 50 + protected_rules: [] + model_dir: repos/vllm-omni/models diff --git a/doc/architecture/SPEC/adapters/base.md b/doc/architecture/SPEC/adapters/base.md index b50ec8ef7..ebf8e5fed 100644 --- a/doc/architecture/SPEC/adapters/base.md +++ b/doc/architecture/SPEC/adapters/base.md @@ -61,3 +61,6 @@ ## 2026-09-28 退役规则不再提供 `render_briefing_docs` 经 `visible_text` 去掉退役规则。 + +## 2026-09-28 knowledge_lifecycle +`knowledge_lifecycle` 加入 `HIGH_RISK_SECTIONS`：知识服务的开关与 `auto_merge` 切换只能经人工审阅修改。 diff --git a/doc/architecture/SPEC/cli.md b/doc/architecture/SPEC/cli.md index fe046f4bb..03085cccd 100644 --- a/doc/architecture/SPEC/cli.md +++ b/doc/architecture/SPEC/cli.md @@ -1,6 +1,6 @@ # cli/ —— 规范 - + `LOC ~1420（6 个文件） · 接口 + 编排门面 · refactor-status: ok` @@ -90,3 +90,6 @@ flag CLI 与 `Copilot` 门面：解析 → 过门 → 执行；并持有 run 目 拆分**已完成**（它曾是内聚拆分候选）。`Copilot` 类完整留在 `copilot.py`， 好让 resolve→execute 的流程能在一个文件里读完；只有 argparse/REPL 前端（`entry.py`） 和两个纯格式化器（`utils.py`）搬了出去。K6（`_gate_and_confirm`）已完成，挂在类上。 + +## 2026-09-28 kb 子命令 +`infermatrix-copilot kb …` 在 argparse 之前分派到 `kb_service.cli.main`（知识服务运维命令），不启动 Copilot/LLM。 diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md new file mode 100644 index 000000000..7d40be03b --- /dev/null +++ b/doc/architecture/SPEC/kb_service.md @@ -0,0 +1,29 @@ +# kb_service/ —— 规范 + + + +`知识服务核心：仓库配置、账本、outbox、CLI · refactor-status: new` + +## 职责 +- `config`：解析 adapter `knowledge_lifecycle`（未知键、非法模式/触发器、缺失审计插件、 + 知识目录与 adapter 不一致、私有上游配 `auto_merge`、`auto_merge` 无校准集均抛 + `LifecycleConfigError`）；`load_registry` 返回按仓库名的注册表，`general` 由服务配置。 +- `ledger`：单个 SQLite，所有表按 `repo` 分区；心跳租约保证单实例；模式变化即代际 +1； + `bump_generation` 使该仓库（或 `*`）之前签发的 outbox 项全部失效，可同时暂停。 +- `outbox`：服务端签发 outbox 项、控制记录与暂停清单，读取发布器签名的回执； + 发布器端 `check_item`（签名、控制记录新鲜度、过期、代际、暂停、私有上游； + 只有 auto_merge 仓库才有非停止类写动作，shadow 与 disabled 一律不发布； + `pause`/`close` 总可执行）； + `verify_holds` 供 kb-gate 使用（过期即失败）。 +- `cli`：`infermatrix-copilot kb keygen|status|pause|resume|control`。 + +## 不变量 +- 模块不出现任何仓库名；仓库差异只来自 adapter 配置。 +- bot 主机无 GitHub 写凭据：服务只写签名的 outbox 项。 +- `knowledge_lifecycle` 属于 adapter 的高风险段，agent 写入被拒。 + +## 依赖（允许） +stdlib + PyYAML + `cryptography`（`kb` extra）+ `.adapters` + `.knowledge_service.signing`。 + +## 测试 +`test_kb_service_core.py`。 diff --git a/doc/architecture/SPEC/knowledge_service.md b/doc/architecture/SPEC/knowledge_service.md index c390ee08c..b5e4a5197 100644 --- a/doc/architecture/SPEC/knowledge_service.md +++ b/doc/architecture/SPEC/knowledge_service.md @@ -30,3 +30,8 @@ from the package, so the kb-gate verifier bundle can vendor them. directory and runs it with a baseline generated for the audited SHA pair: knowledge-document issues are enforced, adapter-baseline maintenance is reported as reconciliation. + +`signing` provides Ed25519 envelopes whose signature covers a purpose tag plus +the canonical JSON payload, so a signature made for one purpose (gate verdict, +outbox item, control record, hold list, publisher ack) never verifies for +another. It needs only `cryptography` and is vendored by the kb-gate verifier. diff --git a/doc/features/knowledge-service.md b/doc/features/knowledge-service.md new file mode 100644 index 000000000..3a60c2823 --- /dev/null +++ b/doc/features/knowledge-service.md @@ -0,0 +1,41 @@ +# 知识服务 —— Copilot 全权负责知识库 + +> **一览** +> | | | +> |---|---| +> | **状态** | 🚧 分阶段落地。已合并：Direct 路由多仓库化与按请求知识视图（#203）、Knowledge Ops API 2.0 与 L1（#204）、generated-baseline 发版审计（#205）；本页随后续 PR 更新 | +> | **做什么** | 知识的创建、维护、退役、删除与发版巡检全部在 Copilot；每次变更经质量门，通过即经合并队列自动合并；ReviewBot 只读消费 | +> | **怎么开关** | 每个仓库在 `adapters//manifest.yaml` 的 `knowledge_lifecycle`（人工审阅的高风险段）中设置 `enabled` 与 `mode: shadow\|auto_merge` | +> | **设计** | 经 GPT-6 sol 评审批准的设计文档（知识库管理重构设计，多仓库，v7） | +> | **硬边界** | bot 主机不持有任何 GitHub 写凭据；所有 GitHub 写动作由 GPU 盒上的发布器以 owner 账号执行，且只执行服务签名、未过期、代际有效的 outbox 项；私有上游仓库只能 shadow，不产生任何公开产物 | + +## 组成 + +| 部分 | 位置 | 职责 | +|---|---|---| +| 仓库配置与注册表 | `kb_service/config.py` | 解析每个 adapter 的 `knowledge_lifecycle`；`general/` 由服务配置 | +| 账本 | `kb_service/ledger.py` | 单个 SQLite（`$KB_STATE_DIR/kb.db`），所有表按 `repo` 分区；单实例租约；代际号 | +| 签名 | `knowledge_service/signing.py` | Ed25519 信封，按用途（判定、outbox、控制记录、暂停清单、回执）隔离签名 | +| outbox | `kb_service/outbox.py` | 服务写签名项、控制记录与暂停清单；发布器执行前用 `check_item` 复核 | +| CLI | `infermatrix-copilot kb …` | `keygen`、`status`、`pause`、`resume`、`control` | + +## 代际与暂停 + +任何暂停、熔断、回滚或切回 `shadow` 都会使该仓库（或全局 `*`）代际 +1，并立即重签 +控制记录与暂停清单。发布器只执行代际与当前一致、未过期（`enqueue`/`post_verdict` +30 分钟，其余 24 小时）且控制记录签发不超过 10 分钟的项；`pause` 与 `close` 总是可执行。 +只有 `auto_merge` 仓库会产生非停止类写动作：`shadow` 只记录“将要做什么”，不开 PR、 +不发评论；模式变化本身也会使代际 +1。 +暂停清单发布在服务的公开 HTTP 端点，kb-gate 在合并队列中读取它，过期或不可达即失败。 + +## 运维 + +```bash +infermatrix-copilot kb keygen --out /etc/infermatrix-kb/service.pem # 公钥提交到 .github/kb-gate.pub +export KB_SIGNING_KEY=/etc/infermatrix-kb/service.pem KB_STATE_DIR=/var/lib/infermatrix-kb +infermatrix-copilot kb status +infermatrix-copilot kb pause --repo vllm-omni --reason "rollback drill" +infermatrix-copilot kb resume --repo vllm-omni +``` + +依赖：`pip install "infermatrix-copilot[kb]"`（cryptography）。 diff --git a/pyproject.toml b/pyproject.toml index fb8f67ad8..409ba2fcf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,6 +17,8 @@ dev = ["pytest>=8.0"] # MCP server (Claude Code / Codex integration) — kept optional so a standalone # CLI install stays dependency-free. mcp = ["mcp>=1.8,<2"] +# Knowledge service (signing, verdict envelopes); the kb-gate verifier pins it by hash. +kb = ["cryptography>=42"] [project.scripts] infermatrix-copilot = "infermatrix_copilot.cli:main" diff --git a/src/infermatrix_copilot/adapters/base.py b/src/infermatrix_copilot/adapters/base.py index e785b4748..98e00b7cf 100644 --- a/src/infermatrix_copilot/adapters/base.py +++ b/src/infermatrix_copilot/adapters/base.py @@ -18,7 +18,7 @@ from ..knowledge_service.lifecycle import visible_text -HIGH_RISK_SECTIONS = ("push", "repo", "upstream", "rebase") +HIGH_RISK_SECTIONS = ("push", "repo", "upstream", "rebase", "knowledge_lifecycle") def expand_path(value: str, extra: dict | None = None) -> str: diff --git a/src/infermatrix_copilot/cli/entry.py b/src/infermatrix_copilot/cli/entry.py index 490746e62..7da81bfbd 100644 --- a/src/infermatrix_copilot/cli/entry.py +++ b/src/infermatrix_copilot/cli/entry.py @@ -6,6 +6,7 @@ from __future__ import annotations import argparse +import sys from ..intent import parse_intents from .copilot import Copilot @@ -136,6 +137,11 @@ def main(argv: list[str] | None = None) -> int: help=argparse.SUPPRESS) parser.add_argument("--execute-strict-reserved", metavar="RUN_ID", help=argparse.SUPPRESS) + raw = list(sys.argv[1:] if argv is None else argv) + if raw and raw[0] == "kb": # the knowledge service has its own subcommands + from ..kb_service.cli import main as kb_main + + return kb_main(raw[1:]) args = parser.parse_args(argv) if args.command == "doctor": # diagnostics only — no Copilot/LLM spin-up diff --git a/src/infermatrix_copilot/kb_service/__init__.py b/src/infermatrix_copilot/kb_service/__init__.py new file mode 100644 index 000000000..d32e65fd1 --- /dev/null +++ b/src/infermatrix_copilot/kb_service/__init__.py @@ -0,0 +1 @@ +"""The knowledge service: owns every change to the knowledge base (see doc/features/knowledge-service.md).""" diff --git a/src/infermatrix_copilot/kb_service/cli.py b/src/infermatrix_copilot/kb_service/cli.py new file mode 100644 index 000000000..ec9910b93 --- /dev/null +++ b/src/infermatrix_copilot/kb_service/cli.py @@ -0,0 +1,145 @@ +"""``infermatrix-copilot kb`` — operate the knowledge service. + + kb keygen --out PATH create a signing key (0600) and print its public key + kb status repositories, modes, generations, queues + kb pause (--repo R | --all) --reason TEXT + kb resume (--repo R | --all) + kb control refresh the signed control record and hold list now + +Pause/resume bump the repository's generation, so every outbox item issued +before it is void; they then re-sign the control record and hold list at once. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +import time +from pathlib import Path + +DEFAULT_STATE_DIR = Path.home() / ".infermatrix-copilot" / "kb" + + +def _state_dir(value: str | None) -> Path: + return Path(value or os.environ.get("KB_STATE_DIR") or DEFAULT_STATE_DIR).expanduser() + + +def _ledger(state_dir: Path): + from .ledger import Ledger + + return Ledger(state_dir / "kb.db") + + +def _registry(): + from ..sdk._resources import adapters_root + from .config import general_lifecycle, load_registry + + general = general_lifecycle( + enabled=os.environ.get("KB_GENERAL_ENABLED", "0") == "1", + mode=os.environ.get("KB_GENERAL_MODE", "shadow"), + ) + return load_registry(Path(os.environ.get("ADAPTERS_DIR") or adapters_root()), general=general) + + +def _signing_key(): + from ..knowledge_service.signing import load_private_key + + path = os.environ.get("KB_SIGNING_KEY") + if not path: + raise SystemExit("KB_SIGNING_KEY is not set (path to the service's Ed25519 private key)") + return load_private_key(path) + + +def _sync_repos(ledger, registry) -> bool: + """Record configured modes; True when any mode changed (and so a + generation was bumped and the signed control record must be re-issued).""" + changed = False + for lifecycle in registry.values(): + changed |= ledger.ensure_repo(lifecycle.repo, lifecycle.mode if lifecycle.enabled else "disabled") + return changed + + +def _outbox(state_dir: Path, ledger): + from .outbox import Outbox + + return Outbox(state_dir, _signing_key(), ledger, clock=time.time) + + +def _public(registry) -> set[str]: + return {name for name, lifecycle in registry.items() if lifecycle.publishes} + + +def _refresh(state_dir: Path, ledger, registry) -> None: + _outbox(state_dir, ledger).transition(lambda: None, public_repos=_public(registry)) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="infermatrix-copilot kb") + parser.add_argument("--state-dir", help="default: $KB_STATE_DIR or ~/.infermatrix-copilot/kb") + sub = parser.add_subparsers(dest="command", required=True) + keygen = sub.add_parser("keygen") + keygen.add_argument("--out", required=True) + sub.add_parser("status") + for name in ("pause", "resume"): + cmd = sub.add_parser(name) + target = cmd.add_mutually_exclusive_group(required=True) + target.add_argument("--repo") + target.add_argument("--all", action="store_true") + if name == "pause": + cmd.add_argument("--reason", required=True) + sub.add_parser("control") + args = parser.parse_args(argv) + + if args.command == "keygen": + from ..knowledge_service.signing import generate_private_key, public_key_text + + key = generate_private_key(args.out) + print(public_key_text(key.public_key())) + return 0 + + state_dir = _state_dir(args.state_dir) + ledger = _ledger(state_dir) + try: + registry = _registry() + if _sync_repos(ledger, registry) and os.environ.get("KB_SIGNING_KEY"): + _refresh(state_dir, ledger, registry) + if args.command == "status": + report = { + "state_dir": str(state_dir), + "active_snapshot": ledger.active_snapshot(), + "repos": [ + {**row, "configured": row["repo"] in registry or row["repo"] == "*", + "open_changesets": len(ledger.changesets(row["repo"], ("open", "pr_open", "signed", "queued"))) + if row["repo"] != "*" else None, + "human_queue": len(ledger.human_queue(row["repo"])) if row["repo"] != "*" else None} + for row in ledger.all_repo_states() + ], + } + json.dump(report, sys.stdout, indent=2, sort_keys=True) + print() + return 0 + if args.command in {"pause", "resume"}: + repo = "*" if args.all else args.repo + if repo != "*" and repo not in registry: + print(f"unknown repository: {repo}", file=sys.stderr) + return 2 + # the state change and its signed publication happen under one + # interprocess lock, so no concurrent refresh can publish a + # pre-pause control record after the pause + if args.command == "pause": + generation = _outbox(state_dir, ledger).transition( + lambda: ledger.bump_generation(repo, pause=True, reason=args.reason), + public_repos=_public(registry)) + else: + generation = _outbox(state_dir, ledger).transition( + lambda: ledger.resume(repo), public_repos=_public(registry)) + print(f"{args.command}d {repo}: generation {generation}") + return 0 + if args.command == "control": + _refresh(state_dir, ledger, registry) + return 0 + finally: + ledger.close() + return 2 diff --git a/src/infermatrix_copilot/kb_service/config.py b/src/infermatrix_copilot/kb_service/config.py new file mode 100644 index 000000000..19a8cf81f --- /dev/null +++ b/src/infermatrix_copilot/kb_service/config.py @@ -0,0 +1,210 @@ +"""Per-repository knowledge-lifecycle configuration and the repository registry. + +Every difference between repositories lives in each adapter's human-reviewed +``knowledge_lifecycle`` manifest section; the service code never names a +repository. ``general`` (the cross-repository slice) is configured by the +service itself because it has no adapter. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from pathlib import Path + +from ..adapters import AdapterError, AdapterRegistry, RepoAdapter + +MODES = ("shadow", "auto_merge") +TRIGGERS = ("github_release", "tag_pattern", "branch_cut", "none") +VISIBILITIES = ("public", "private") +GENERAL = "general" +_NAME = re.compile(r"[a-z0-9][a-z0-9._-]*") +_FULL_NAME = re.compile(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+") +_KEYS = { + "enabled", "mode", "knowledge_dir", "intake", "release", "sweep", + "upstream_visibility", "circuit_breaker", "protected_rules", + "calibration_set", "model_dir", +} + + +class LifecycleConfigError(ValueError): + """An adapter's knowledge_lifecycle section is invalid.""" + + +@dataclass(frozen=True) +class IntakeConfig: + merged_prs: bool = True + copilot_runs: bool = True + human_prs: bool = True + daily_model_budget_usd: float = 0.0 + + +@dataclass(frozen=True) +class ReleaseConfig: + trigger: str = "none" + tag_pattern: str = "" + auditor: str = "" # plugin path relative to the adapter directory + + +@dataclass(frozen=True) +class RepoLifecycle: + repo: str # knowledge repo name, or "general" + full_name: str # upstream owner/name ("" for general) + enabled: bool + mode: str + knowledge_dir: str # "repos/" or "general" + intake: IntakeConfig = field(default_factory=IntakeConfig) + release: ReleaseConfig = field(default_factory=ReleaseConfig) + fallback_interval_days: int = 30 + upstream_visibility: str = "public" + retire_ratio: float = 0.10 + max_files: int = 50 + protected_rules: tuple[str, ...] = () + calibration_set: str = "" + model_dir: str = "" + adapter_dir: Path | None = None + + @property + def auto_merge(self) -> bool: + return self.enabled and self.mode == "auto_merge" + + @property + def publishes(self) -> bool: + """Whether anything about this repository may appear in the PUBLIC + knowledge repository (PRs, verdict comments, reports).""" + return self.upstream_visibility == "public" + + +def _bool(value: object, where: str) -> bool: + if not isinstance(value, bool): + raise LifecycleConfigError(f"{where} must be true or false") + return value + + +def _mapping(value: object, where: str, allowed: set[str]) -> dict: + if value is None: + return {} + if not isinstance(value, dict): + raise LifecycleConfigError(f"{where} must be a mapping") + unknown = set(value) - allowed + if unknown: + raise LifecycleConfigError(f"{where} has unknown keys {sorted(unknown)}") + return value + + +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") + if section is None: + return None + where = f"adapters/{adapter.name}/manifest.yaml knowledge_lifecycle" + data = _mapping(section, where, _KEYS) + repo_meta = adapter.manifest.get("repo") or {} + knowledge = adapter.manifest.get("knowledge") or {} + repo_subdir = str(knowledge.get("repo_subdir") or "").strip("/") + if not repo_subdir.startswith("repos/"): + raise LifecycleConfigError(f"{where}: adapter knowledge.repo_subdir must be repos/") + repo = repo_subdir.removeprefix("repos/") + if not _NAME.fullmatch(repo): + raise LifecycleConfigError(f"{where}: invalid repo name {repo!r}") + full_name = str(repo_meta.get("full_name") or "") + if not _FULL_NAME.fullmatch(full_name): + raise LifecycleConfigError(f"{where}: adapter repo.full_name must be owner/name") + enabled = _bool(data.get("enabled", False), f"{where}.enabled") + mode = str(data.get("mode", "shadow")) + if mode not in MODES: + raise LifecycleConfigError(f"{where}.mode must be one of {MODES}") + knowledge_dir = str(data.get("knowledge_dir", repo_subdir)).strip("/") + if knowledge_dir != repo_subdir: + raise LifecycleConfigError(f"{where}.knowledge_dir must equal the adapter repo_subdir {repo_subdir}") + intake = _mapping(data.get("intake"), f"{where}.intake", + {"merged_prs", "copilot_runs", "human_prs", "daily_model_budget_usd"}) + release = _mapping(data.get("release"), f"{where}.release", {"trigger", "tag_pattern", "auditor"}) + sweep = _mapping(data.get("sweep"), f"{where}.sweep", {"fallback_interval_days"}) + breaker = _mapping(data.get("circuit_breaker"), f"{where}.circuit_breaker", + {"retire_ratio", "max_files"}) + trigger = str(release.get("trigger", "none")) + if trigger not in TRIGGERS: + raise LifecycleConfigError(f"{where}.release.trigger must be one of {TRIGGERS}") + if trigger == "tag_pattern" and not release.get("tag_pattern"): + raise LifecycleConfigError(f"{where}.release.tag_pattern is required for tag_pattern") + auditor = str(release.get("auditor") or "") + if auditor and not (Path(adapter.root) / auditor).is_file(): + raise LifecycleConfigError(f"{where}.release.auditor does not exist: {auditor}") + visibility = str(data.get("upstream_visibility", "public")) + if visibility not in VISIBILITIES: + raise LifecycleConfigError(f"{where}.upstream_visibility must be one of {VISIBILITIES}") + if visibility == "private" and mode == "auto_merge": + # the knowledge repository is public: a private upstream may never publish + raise LifecycleConfigError(f"{where}: a private upstream can only run in shadow mode") + retire_ratio = float(breaker.get("retire_ratio", 0.10)) + max_files = int(breaker.get("max_files", 50)) + if not 0 < retire_ratio <= 1 or max_files < 1: + raise LifecycleConfigError(f"{where}.circuit_breaker values are out of range") + budget = float(intake.get("daily_model_budget_usd", 0.0)) + if budget < 0: + raise LifecycleConfigError(f"{where}.intake.daily_model_budget_usd must be >= 0") + protected = data.get("protected_rules") or [] + if not isinstance(protected, list) or any(not isinstance(item, str) for item in protected): + raise LifecycleConfigError(f"{where}.protected_rules must be a list of rule IDs") + if mode == "auto_merge" and not data.get("calibration_set"): + raise LifecycleConfigError(f"{where}: auto_merge requires a calibration_set") + return RepoLifecycle( + repo=repo, + full_name=full_name, + enabled=enabled, + mode=mode, + knowledge_dir=knowledge_dir, + intake=IntakeConfig( + merged_prs=_bool(intake.get("merged_prs", True), f"{where}.intake.merged_prs"), + copilot_runs=_bool(intake.get("copilot_runs", True), f"{where}.intake.copilot_runs"), + human_prs=_bool(intake.get("human_prs", True), f"{where}.intake.human_prs"), + daily_model_budget_usd=budget, + ), + release=ReleaseConfig(trigger=trigger, tag_pattern=str(release.get("tag_pattern") or ""), + auditor=auditor), + fallback_interval_days=int(sweep.get("fallback_interval_days", 30)), + upstream_visibility=visibility, + retire_ratio=retire_ratio, + max_files=max_files, + protected_rules=tuple(protected), + calibration_set=str(data.get("calibration_set") or ""), + model_dir=str(data.get("model_dir") or ""), + adapter_dir=Path(adapter.root), + ) + + +def general_lifecycle(*, enabled: bool, mode: str = "shadow") -> RepoLifecycle: + """The cross-repository slice: sources are Copilot runs and human PRs only, + no release events, monthly full sweep.""" + if mode not in MODES: + raise LifecycleConfigError(f"general mode must be one of {MODES}") + return RepoLifecycle( + repo=GENERAL, full_name="", enabled=enabled, mode=mode, knowledge_dir=GENERAL, + intake=IntakeConfig(merged_prs=False), + release=ReleaseConfig(trigger="none"), + ) + + +def load_registry(adapters_dir: str | Path, *, general: RepoLifecycle | None = None + ) -> dict[str, RepoLifecycle]: + """Every adapter with a knowledge_lifecycle section, keyed by repo name. + + An invalid section raises: a misconfigured repository must never run with + guessed settings. Adapters without the section are simply not served. + """ + registry: dict[str, RepoLifecycle] = {} + try: + adapters = AdapterRegistry(adapters_dir).all() + except AdapterError as exc: + raise LifecycleConfigError(f"adapters cannot be loaded: {exc}") from exc + for adapter in adapters: + lifecycle = parse_lifecycle(adapter) + if lifecycle is None: + continue + if lifecycle.repo in registry: + raise LifecycleConfigError(f"two adapters claim knowledge repo {lifecycle.repo}") + registry[lifecycle.repo] = lifecycle + if general is not None: + registry[GENERAL] = general + return registry diff --git a/src/infermatrix_copilot/kb_service/ledger.py b/src/infermatrix_copilot/kb_service/ledger.py new file mode 100644 index 000000000..490fc4968 --- /dev/null +++ b/src/infermatrix_copilot/kb_service/ledger.py @@ -0,0 +1,392 @@ +"""The knowledge service's ledger: one SQLite file, every table keyed by repo. + +It is the single source of truth for knowledge events, candidate operations, +gate verdicts, change sets, activations, retirements, sweeps, the human queue, +outbox items and scheduler cursors. A repository's failure or pause never +touches another repository's rows. + +Only one service instance may run: ``acquire_lease`` takes a heartbeat lease +row inside a write transaction; a stale lease (heartbeat older than its TTL) can +be taken over, a live one cannot. +""" + +from __future__ import annotations + +import json +import os +import sqlite3 +import time +import uuid +from contextlib import contextmanager +from pathlib import Path +from typing import Any, Iterator + +SCHEMA_VERSION = 1 +GLOBAL = "*" + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT NOT NULL); +CREATE TABLE IF NOT EXISTS repo_state ( + repo TEXT PRIMARY KEY, + generation INTEGER NOT NULL DEFAULT 1, + mode TEXT NOT NULL DEFAULT 'shadow', + paused INTEGER NOT NULL DEFAULT 0, + pause_reason TEXT NOT NULL DEFAULT '', + updated_at REAL NOT NULL +); +CREATE TABLE IF NOT EXISTS cursors ( + repo TEXT NOT NULL, name TEXT NOT NULL, value TEXT NOT NULL, updated_at REAL NOT NULL, + PRIMARY KEY (repo, name) +); +CREATE TABLE IF NOT EXISTS events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + repo TEXT NOT NULL, source TEXT NOT NULL, external_id TEXT NOT NULL, + payload TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending', + attempts INTEGER NOT NULL DEFAULT 0, detail TEXT NOT NULL DEFAULT '', + created_at REAL NOT NULL, updated_at REAL NOT NULL, + UNIQUE (repo, source, external_id) +); +CREATE TABLE IF NOT EXISTS candidates ( + id TEXT PRIMARY KEY, repo TEXT NOT NULL, event_id INTEGER, + changeset_id TEXT, operation TEXT NOT NULL, content_sha256 TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', created_at REAL NOT NULL, updated_at REAL NOT NULL +); +CREATE TABLE IF NOT EXISTS verdicts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, repo TEXT NOT NULL, + changeset_id TEXT, block_id TEXT NOT NULL DEFAULT '', layer TEXT NOT NULL, + verdict TEXT NOT NULL, model TEXT NOT NULL DEFAULT '', model_version TEXT NOT NULL DEFAULT '', + detail TEXT NOT NULL DEFAULT '{}', created_at REAL NOT NULL +); +CREATE TABLE IF NOT EXISTS changesets ( + id TEXT PRIMARY KEY, repo TEXT NOT NULL, kind TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'open', generation INTEGER NOT NULL, + branch TEXT NOT NULL DEFAULT '', pr_number INTEGER, head_sha TEXT NOT NULL DEFAULT '', + merge_sha TEXT NOT NULL DEFAULT '', detail TEXT NOT NULL DEFAULT '{}', + created_at REAL NOT NULL, updated_at REAL NOT NULL +); +CREATE TABLE IF NOT EXISTS activations ( + id INTEGER PRIMARY KEY AUTOINCREMENT, snapshot TEXT NOT NULL, previous TEXT NOT NULL DEFAULT '', + activated_at REAL NOT NULL, detail TEXT NOT NULL DEFAULT '{}' +); +CREATE TABLE IF NOT EXISTS retirements ( + repo TEXT NOT NULL, rule_id TEXT NOT NULL, page TEXT NOT NULL, + retired_at_release TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'retired', + updated_at REAL NOT NULL, PRIMARY KEY (repo, rule_id) +); +CREATE TABLE IF NOT EXISTS sweeps ( + id TEXT PRIMARY KEY, repo TEXT NOT NULL, from_sha TEXT NOT NULL, to_sha TEXT NOT NULL, + tier TEXT NOT NULL, status TEXT NOT NULL, report TEXT NOT NULL DEFAULT '{}', + started_at REAL NOT NULL, finished_at REAL +); +CREATE TABLE IF NOT EXISTS human_queue ( + id INTEGER PRIMARY KEY AUTOINCREMENT, repo TEXT NOT NULL, changeset_id TEXT, + reason TEXT NOT NULL, created_at REAL NOT NULL, resolved_at REAL +); +CREATE TABLE IF NOT EXISTS outbox ( + id TEXT PRIMARY KEY, repo TEXT NOT NULL, kind TEXT NOT NULL, generation INTEGER NOT NULL, + expires_at REAL NOT NULL, status TEXT NOT NULL DEFAULT 'written', + payload_sha256 TEXT NOT NULL, ack TEXT NOT NULL DEFAULT '', + created_at REAL NOT NULL, acked_at REAL +); +CREATE INDEX IF NOT EXISTS events_by_status ON events (repo, status); +CREATE INDEX IF NOT EXISTS changesets_by_status ON changesets (repo, status); +CREATE INDEX IF NOT EXISTS outbox_by_status ON outbox (repo, status); +""" + + +class LeaseError(RuntimeError): + """Another live knowledge-service instance holds the lease.""" + + +class Ledger: + def __init__(self, path: str | Path, *, clock=time.time): + self.path = Path(path) + self.path.parent.mkdir(parents=True, exist_ok=True) + self._clock = clock + self._conn = sqlite3.connect(self.path, timeout=30, isolation_level=None) + self._conn.row_factory = sqlite3.Row + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.execute("PRAGMA foreign_keys=ON") + self._conn.executescript(_SCHEMA) + with self.tx() as cur: + row = cur.execute("SELECT value FROM meta WHERE key='schema_version'").fetchone() + if row is None: + cur.execute("INSERT INTO meta VALUES ('schema_version', ?)", (str(SCHEMA_VERSION),)) + elif int(row["value"]) != SCHEMA_VERSION: + raise RuntimeError(f"ledger schema {row['value']} != {SCHEMA_VERSION}") + try: + os.chmod(self.path, 0o600) + except OSError: + pass + + def close(self) -> None: + self._conn.close() + + @contextmanager + def tx(self) -> Iterator[sqlite3.Cursor]: + cur = self._conn.cursor() + cur.execute("BEGIN IMMEDIATE") + try: + yield cur + except BaseException: + cur.execute("ROLLBACK") + raise + else: + cur.execute("COMMIT") + + # -- lease --------------------------------------------------------------- + + def acquire_lease(self, owner: str | None = None, *, ttl: float = 120.0) -> str: + owner = owner or f"{os.uname().nodename}:{os.getpid()}:{uuid.uuid4().hex[:8]}" + now = self._clock() + with self.tx() as cur: + row = cur.execute("SELECT value FROM meta WHERE key='lease'").fetchone() + if row is not None: + held = json.loads(row["value"]) + if held["owner"] != owner and now - held["heartbeat"] < held["ttl"]: + raise LeaseError(f"lease held by {held['owner']}") + cur.execute( + "INSERT INTO meta VALUES ('lease', ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value", + (json.dumps({"owner": owner, "heartbeat": now, "ttl": ttl}),), + ) + return owner + + def heartbeat(self, owner: str) -> None: + with self.tx() as cur: + row = cur.execute("SELECT value FROM meta WHERE key='lease'").fetchone() + held = json.loads(row["value"]) if row else None + if held is None or held["owner"] != owner: + raise LeaseError("lease was lost") + held["heartbeat"] = self._clock() + cur.execute("UPDATE meta SET value=? WHERE key='lease'", (json.dumps(held),)) + + def release_lease(self, owner: str) -> None: + with self.tx() as cur: + row = cur.execute("SELECT value FROM meta WHERE key='lease'").fetchone() + if row and json.loads(row["value"])["owner"] == owner: + cur.execute("DELETE FROM meta WHERE key='lease'") + + # -- repo state and generations ----------------------------------------- + + def ensure_repo(self, repo: str, mode: str) -> bool: + """Register ``repo`` or record its configured mode. A CHANGE of mode + bumps the generation in the same transaction, so nothing issued under + the previous mode (e.g. an enqueue before a switch to shadow) survives + a round trip back. Returns whether the generation changed.""" + now = self._clock() + with self.tx() as cur: + row = cur.execute("SELECT mode FROM repo_state WHERE repo=?", (repo,)).fetchone() + changed = row is not None and row["mode"] != mode + if row is None: + cur.execute("INSERT INTO repo_state (repo, mode, updated_at) VALUES (?, ?, ?)", + (repo, mode, now)) + elif changed: + cur.execute( + "UPDATE repo_state SET mode=?, generation=generation+1, updated_at=? WHERE repo=?", + (mode, now, repo)) + cur.execute( + "INSERT OR IGNORE INTO repo_state (repo, mode, updated_at) VALUES (?, 'global', ?)", + (GLOBAL, now), + ) + return changed + + def repo_state(self, repo: str) -> dict[str, Any]: + row = self._conn.execute("SELECT * FROM repo_state WHERE repo=?", (repo,)).fetchone() + if row is None: + raise KeyError(f"unknown repo in ledger: {repo}") + return dict(row) + + def all_repo_states(self) -> list[dict[str, Any]]: + return [dict(r) for r in self._conn.execute("SELECT * FROM repo_state ORDER BY repo")] + + def bump_generation(self, repo: str, *, pause: bool, reason: str) -> int: + """Invalidate every outstanding outbox item of ``repo`` (or of all repos + for ``GLOBAL``) and optionally pause it. Returns the new generation.""" + with self.tx() as cur: + row = cur.execute("SELECT generation FROM repo_state WHERE repo=?", (repo,)).fetchone() + if row is None: + raise KeyError(f"unknown repo in ledger: {repo}") + generation = int(row["generation"]) + 1 + cur.execute( + "UPDATE repo_state SET generation=?, paused=?, pause_reason=?, updated_at=? WHERE repo=?", + (generation, 1 if pause else 0, reason if pause else "", self._clock(), repo), + ) + return generation + + def resume(self, repo: str) -> int: + """Unpause; bumps the generation too, so nothing issued while paused revives.""" + return self.bump_generation(repo, pause=False, reason="") + + # -- cursors ------------------------------------------------------------- + + def get_cursor(self, repo: str, name: str) -> str | None: + row = self._conn.execute( + "SELECT value FROM cursors WHERE repo=? AND name=?", (repo, name)).fetchone() + return row["value"] if row else None + + def set_cursor(self, repo: str, name: str, value: str) -> None: + with self.tx() as cur: + cur.execute( + "INSERT INTO cursors VALUES (?, ?, ?, ?) ON CONFLICT(repo, name) " + "DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at", + (repo, name, value, self._clock()), + ) + + # -- events -------------------------------------------------------------- + + def record_event(self, repo: str, source: str, external_id: str, payload: dict) -> int | None: + """Insert an event once (idempotent by repo/source/external_id).""" + now = self._clock() + with self.tx() as cur: + cur.execute( + "INSERT OR IGNORE INTO events (repo, source, external_id, payload, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?)", + (repo, source, external_id, json.dumps(payload, sort_keys=True), now, now), + ) + if cur.rowcount == 0: + return None + return int(cur.lastrowid) + + def events(self, repo: str, status: str, *, limit: int = 100) -> list[dict[str, Any]]: + rows = self._conn.execute( + "SELECT * FROM events WHERE repo=? AND status=? ORDER BY id LIMIT ?", + (repo, status, limit)).fetchall() + return [{**dict(r), "payload": json.loads(r["payload"])} for r in rows] + + def set_event_status(self, event_id: int, status: str, detail: str = "") -> None: + with self.tx() as cur: + cur.execute( + "UPDATE events SET status=?, detail=?, attempts=attempts+?, updated_at=? WHERE id=?", + (status, detail, 1 if status == "error" else 0, self._clock(), event_id), + ) + + # -- change sets, verdicts, queue ---------------------------------------- + + def create_changeset(self, repo: str, kind: str, *, detail: dict | None = None) -> str: + changeset_id = f"{repo}-{kind}-{uuid.uuid4().hex[:12]}" + generation = self.repo_state(repo)["generation"] + now = self._clock() + with self.tx() as cur: + cur.execute( + "INSERT INTO changesets (id, repo, kind, generation, detail, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + (changeset_id, repo, kind, generation, json.dumps(detail or {}, sort_keys=True), now, now), + ) + return changeset_id + + def update_changeset(self, changeset_id: str, **fields: Any) -> None: + allowed = {"status", "branch", "pr_number", "head_sha", "merge_sha", "detail"} + unknown = set(fields) - allowed + if unknown: + raise ValueError(f"unknown changeset fields: {sorted(unknown)}") + if "detail" in fields: + fields["detail"] = json.dumps(fields["detail"], sort_keys=True) + assignments = ", ".join(f"{key}=?" for key in fields) + with self.tx() as cur: + cur.execute( + f"UPDATE changesets SET {assignments}, updated_at=? WHERE id=?", + (*fields.values(), self._clock(), changeset_id), + ) + + def changeset(self, changeset_id: str) -> dict[str, Any]: + row = self._conn.execute("SELECT * FROM changesets WHERE id=?", (changeset_id,)).fetchone() + if row is None: + raise KeyError(changeset_id) + return {**dict(row), "detail": json.loads(row["detail"])} + + def changesets(self, repo: str, statuses: tuple[str, ...]) -> list[dict[str, Any]]: + marks = ",".join("?" for _ in statuses) + rows = self._conn.execute( + f"SELECT * FROM changesets WHERE repo=? AND status IN ({marks}) ORDER BY created_at", + (repo, *statuses)).fetchall() + return [{**dict(r), "detail": json.loads(r["detail"])} for r in rows] + + def record_verdict(self, repo: str, *, layer: str, verdict: str, changeset_id: str | None = None, + block_id: str = "", model: str = "", model_version: str = "", + detail: dict | None = None) -> None: + with self.tx() as cur: + cur.execute( + "INSERT INTO verdicts (repo, changeset_id, block_id, layer, verdict, model, model_version, " + "detail, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + (repo, changeset_id, block_id, layer, verdict, model, model_version, + json.dumps(detail or {}, sort_keys=True), self._clock()), + ) + + def verdicts(self, changeset_id: str) -> list[dict[str, Any]]: + rows = self._conn.execute( + "SELECT * FROM verdicts WHERE changeset_id=? ORDER BY id", (changeset_id,)).fetchall() + return [{**dict(r), "detail": json.loads(r["detail"])} for r in rows] + + def enqueue_human(self, repo: str, reason: str, changeset_id: str | None = None) -> None: + with self.tx() as cur: + cur.execute( + "INSERT INTO human_queue (repo, changeset_id, reason, created_at) VALUES (?, ?, ?, ?)", + (repo, changeset_id, reason, self._clock()), + ) + + def human_queue(self, repo: str | None = None) -> list[dict[str, Any]]: + sql = "SELECT * FROM human_queue WHERE resolved_at IS NULL" + args: tuple = () + if repo is not None: + sql += " AND repo=?" + args = (repo,) + return [dict(r) for r in self._conn.execute(sql + " ORDER BY id", args)] + + # -- retirements and activations ------------------------------------------ + + def record_retirement(self, repo: str, rule_id: str, page: str, release: str) -> None: + with self.tx() as cur: + cur.execute( + "INSERT INTO retirements VALUES (?, ?, ?, ?, 'retired', ?) ON CONFLICT(repo, rule_id) " + "DO UPDATE SET page=excluded.page, retired_at_release=excluded.retired_at_release, " + "status='retired', updated_at=excluded.updated_at", + (repo, rule_id, page, release, self._clock()), + ) + + def purge_eligible(self, repo: str, current_release: str) -> list[dict[str, Any]]: + rows = self._conn.execute( + "SELECT * FROM retirements WHERE repo=? AND status='retired' AND retired_at_release<>?", + (repo, current_release)).fetchall() + return [dict(r) for r in rows] + + def mark_purged(self, repo: str, rule_id: str) -> None: + with self.tx() as cur: + cur.execute("UPDATE retirements SET status='purged', updated_at=? WHERE repo=? AND rule_id=?", + (self._clock(), repo, rule_id)) + + def record_activation(self, snapshot: str, detail: dict | None = None) -> None: + previous = self.active_snapshot() or "" + with self.tx() as cur: + cur.execute( + "INSERT INTO activations (snapshot, previous, activated_at, detail) VALUES (?, ?, ?, ?)", + (snapshot, previous, self._clock(), json.dumps(detail or {}, sort_keys=True)), + ) + + def active_snapshot(self) -> str | None: + row = self._conn.execute( + "SELECT snapshot FROM activations ORDER BY id DESC LIMIT 1").fetchone() + return row["snapshot"] if row else None + + # -- outbox bookkeeping --------------------------------------------------- + + def record_outbox(self, item_id: str, repo: str, kind: str, generation: int, + expires_at: float, payload_sha256: str) -> None: + with self.tx() as cur: + cur.execute( + "INSERT INTO outbox (id, repo, kind, generation, expires_at, payload_sha256, created_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + (item_id, repo, kind, generation, expires_at, payload_sha256, self._clock()), + ) + + def ack_outbox(self, item_id: str, status: str, ack: dict) -> None: + if status not in {"acked", "rejected"}: + raise ValueError("ack status must be acked or rejected") + with self.tx() as cur: + cur.execute( + "UPDATE outbox SET status=?, ack=?, acked_at=? WHERE id=? AND status='written'", + (status, json.dumps(ack, sort_keys=True), self._clock(), item_id), + ) + + def outbox_items(self, repo: str, status: str = "written") -> list[dict[str, Any]]: + return [dict(r) for r in self._conn.execute( + "SELECT * FROM outbox WHERE repo=? AND status=? ORDER BY created_at", (repo, status))] diff --git a/src/infermatrix_copilot/kb_service/outbox.py b/src/infermatrix_copilot/kb_service/outbox.py new file mode 100644 index 000000000..a8c7ee389 --- /dev/null +++ b/src/infermatrix_copilot/kb_service/outbox.py @@ -0,0 +1,283 @@ +"""The signed hand-off between the knowledge service and the GPU-box publisher. + +The service holds no GitHub write credential. Every GitHub write is an outbox +item the service signs; the publisher (on another host, with the owner's ``gh`` +login) pulls items over SSH, re-checks them and performs them. Layout under the +service state directory:: + + outbox/.json signed item (purpose kb-outbox-item) + outbox/control.json signed control record, refreshed every minute + inbox/acks/.json the publisher's result for an item + public/holds.json signed hold list, served read-only over HTTP; the + kb-gate verifier fails knowledge segments it names + +Staleness is enforced on both sides. Each item carries the repository's and the +global generation at issue time and an expiry; any pause, circuit break, +rollback or switch to shadow bumps a generation, so everything issued before it +is void when the publisher next runs (``pause`` and ``close`` excepted: they only +ever stop things). The publisher also refuses to act on a control record older +than ``CONTROL_MAX_AGE``. +""" + +from __future__ import annotations + +import fcntl +import hashlib +import json +import os +import tempfile +import uuid +from contextlib import contextmanager +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Iterator + +from ..knowledge_service.signing import canonical_json, sign, verify + +ITEM_KINDS = ( + "open_pr", "open_companion_pr", "post_verdict", "enqueue", + "update_branch", "pause", "close", +) +ALWAYS_EXECUTABLE = frozenset({"pause", "close"}) +ITEM_TTL = { + "open_pr": 24 * 3600, "open_companion_pr": 24 * 3600, + "post_verdict": 30 * 60, "enqueue": 30 * 60, + "update_branch": 24 * 3600, "pause": 24 * 3600, "close": 24 * 3600, +} +CONTROL_MAX_AGE = 10 * 60 +HOLDS_MAX_AGE = 10 * 60 + + +class OutboxError(RuntimeError): + """An item is invalid, stale, or must not be executed.""" + + +def atomic_write_json(path: Path, data: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) + try: + with os.fdopen(fd, "wb") as handle: + handle.write(canonical_json(data)) + handle.flush() + os.fsync(handle.fileno()) + os.chmod(tmp, 0o644) + os.replace(tmp, path) + except BaseException: + try: + os.unlink(tmp) + except FileNotFoundError: + pass + raise + + +@dataclass(frozen=True) +class OutboxItem: + id: str + kind: str + repo: str + generation: int + global_generation: int + issued_at: float + expires_at: float + body: dict + + def to_payload(self) -> dict: + return { + "id": self.id, "kind": self.kind, "repo": self.repo, + "generation": self.generation, "global_generation": self.global_generation, + "issued_at": self.issued_at, "expires_at": self.expires_at, "body": self.body, + } + + @classmethod + def from_payload(cls, payload: Any) -> "OutboxItem": + if not isinstance(payload, dict) or set(payload) != { + "id", "kind", "repo", "generation", "global_generation", "issued_at", "expires_at", "body", + }: + raise OutboxError("outbox item payload has the wrong fields") + if payload["kind"] not in ITEM_KINDS: + raise OutboxError(f"unknown outbox item kind: {payload['kind']}") + if not isinstance(payload["body"], dict): + raise OutboxError("outbox item body must be a mapping") + return cls( + id=str(payload["id"]), kind=str(payload["kind"]), repo=str(payload["repo"]), + generation=int(payload["generation"]), global_generation=int(payload["global_generation"]), + issued_at=float(payload["issued_at"]), expires_at=float(payload["expires_at"]), + body=payload["body"], + ) + + +class Outbox: + """Service side: write signed items, control records, holds; read acks.""" + + def __init__(self, state_dir: str | Path, key, ledger, *, clock): + self.root = Path(state_dir) + self._key = key + self._ledger = ledger + self._clock = clock + self._lock_depth = 0 + self._lock_fd: int | None = None + + @contextmanager + def publication_lock(self) -> Iterator[None]: + """Interprocess lock around every "read ledger state -> sign -> write" + publication, and around state transitions that must be published + atomically (pause/resume). Without it a refresh that read the state + before a pause could overwrite the pause's newer control record.""" + if self._lock_depth: + self._lock_depth += 1 + try: + yield + finally: + self._lock_depth -= 1 + return + self.root.mkdir(parents=True, exist_ok=True) + fd = os.open(self.root / ".publish.lock", os.O_RDWR | os.O_CREAT, 0o600) + try: + fcntl.flock(fd, fcntl.LOCK_EX) + self._lock_fd, self._lock_depth = fd, 1 + try: + yield + finally: + self._lock_depth, self._lock_fd = 0, None + fcntl.flock(fd, fcntl.LOCK_UN) + finally: + os.close(fd) + + @property + def outbox_dir(self) -> Path: + return self.root / "outbox" + + @property + def acks_dir(self) -> Path: + return self.root / "inbox" / "acks" + + @property + def holds_path(self) -> Path: + return self.root / "public" / "holds.json" + + def issue(self, repo: str, kind: str, body: dict) -> OutboxItem: + if kind not in ITEM_KINDS: + raise OutboxError(f"unknown outbox item kind: {kind}") + state = self._ledger.repo_state(repo) + now = self._clock() + item = OutboxItem( + id=f"{int(now)}-{uuid.uuid4().hex[:12]}", kind=kind, repo=repo, + generation=int(state["generation"]), + global_generation=int(self._ledger.repo_state("*")["generation"]), + issued_at=now, expires_at=now + ITEM_TTL[kind], body=body, + ) + envelope = sign("kb-outbox-item", item.to_payload(), self._key) + digest = hashlib.sha256(canonical_json(envelope)).hexdigest() + self._ledger.record_outbox(item.id, repo, kind, item.generation, item.expires_at, digest) + atomic_write_json(self.outbox_dir / f"{item.id}.json", envelope) + return item + + def _control_payload(self) -> dict: + repos = { + row["repo"]: {"generation": int(row["generation"]), "mode": row["mode"], + "paused": bool(row["paused"])} + for row in self._ledger.all_repo_states() + } + return {"issued_at": self._clock(), "repos": repos} + + def refresh_control(self) -> dict: + with self.publication_lock(): + payload = self._control_payload() + atomic_write_json(self.outbox_dir / "control.json", sign("kb-control", payload, self._key)) + return payload + + def transition(self, change, *, public_repos: set[str]) -> Any: + """Apply a ledger state change (pause/resume/mode) and republish the + control record and hold list, all under the publication lock.""" + with self.publication_lock(): + result = change() + self.refresh_control() + sequence = int(self._ledger.get_cursor("*", "holds_sequence") or 0) + 1 + self._ledger.set_cursor("*", "holds_sequence", str(sequence)) + self.publish_holds(sequence=sequence, public_repos=public_repos) + return result + + def publish_holds(self, *, sequence: int, public_repos: set[str]) -> dict: + """Signed hold list: paused repositories and PRs whose change sets are + paused. It is served publicly, so ONLY repositories in ``public_repos`` + (public upstreams) may appear; a private upstream's name, pause state + or PRs never do (it publishes nothing, so it has no PR to hold).""" + with self.publication_lock(): + return self._publish_holds(sequence, frozenset(public_repos)) + + def _publish_holds(self, sequence: int, public_repos: frozenset[str]) -> dict: + states = [row for row in self._ledger.all_repo_states() + if row["repo"] == "*" or row["repo"] in public_repos] + paused = [row["repo"] for row in states if row["paused"] and row["repo"] != "*"] + global_hold = any(row["repo"] == "*" and row["paused"] for row in states) + prs = sorted({ + int(cs["pr_number"]) + for row in states if row["repo"] != "*" + for cs in self._ledger.changesets(row["repo"], ("paused",)) + if cs["pr_number"] is not None + }) + payload = {"issued_at": self._clock(), "sequence": sequence, "global": global_hold, + "repos": sorted(paused), "prs": prs} + atomic_write_json(self.holds_path, sign("kb-holds", payload, self._key)) + return payload + + def collect_acks(self, public_key) -> list[dict]: + """Read publisher acks (signed by the PUBLISHER's key), record them, remove the files.""" + results = [] + if not self.acks_dir.is_dir(): + return results + for path in sorted(self.acks_dir.glob("*.json")): + try: + payload = verify("kb-ack", json.loads(path.read_text(encoding="utf-8")), public_key) + item_id = str(payload["item_id"]) + status = "acked" if payload.get("ok") else "rejected" + self._ledger.ack_outbox(item_id, status, payload) + results.append(payload) + except Exception as exc: # malformed or forged ack: keep for inspection + path.rename(path.with_suffix(".invalid")) + results.append({"invalid": path.name, "error": str(exc)}) + continue + path.unlink() + (self.outbox_dir / f"{item_id}.json").unlink(missing_ok=True) + return results + + +# -- publisher side -------------------------------------------------------------- + +def check_item(envelope: Any, control_envelope: Any, public_key, *, now: float, + repo_publishes: bool, repo_auto_merge: bool) -> OutboxItem: + """What the publisher runs before ANY write. Raises ``OutboxError`` if the + item must not be executed; returns the verified item otherwise.""" + control = verify("kb-control", control_envelope, public_key) + if now - float(control["issued_at"]) > CONTROL_MAX_AGE: + raise OutboxError("control record is stale; the service may be down — doing nothing") + item = OutboxItem.from_payload(verify("kb-outbox-item", envelope, public_key)) + repos = control["repos"] + if item.repo not in repos: + raise OutboxError(f"control record does not know repo {item.repo}") + if item.kind in ALWAYS_EXECUTABLE: + return item + if now > item.expires_at: + raise OutboxError("item expired") + current = repos[item.repo] + if item.generation != current["generation"] or item.global_generation != repos["*"]["generation"]: + raise OutboxError("item was issued before a pause, rollback or mode change (stale generation)") + if current["paused"] or repos["*"]["paused"]: + raise OutboxError("repository is paused") + if not repo_publishes: + raise OutboxError("repository has a private upstream: nothing may be published") + # shadow records what it WOULD do and publishes nothing; a disabled repo + # does nothing. Only an auto_merge repository gets non-stop GitHub writes, + # and both the signed control record and the publisher's own copy of the + # adapter configuration must say so. + if current["mode"] != "auto_merge" or not repo_auto_merge: + raise OutboxError(f"repository is not in auto_merge mode (control: {current['mode']})") + return item + + +def verify_holds(envelope: Any, public_key, *, now: float) -> dict: + """Used by the kb-gate verifier: a stale or unverifiable hold list fails closed.""" + holds = verify("kb-holds", envelope, public_key) + if now - float(holds["issued_at"]) > HOLDS_MAX_AGE: + raise OutboxError("hold list is stale") + return holds diff --git a/src/infermatrix_copilot/knowledge_service/signing.py b/src/infermatrix_copilot/knowledge_service/signing.py new file mode 100644 index 000000000..454b1a72a --- /dev/null +++ b/src/infermatrix_copilot/knowledge_service/signing.py @@ -0,0 +1,117 @@ +"""Ed25519 envelopes for knowledge-gate verdicts, outbox items and control records. + +An envelope is ``{"payload": , "key_id": , "signature": }``. +The signature covers the canonical JSON of the payload (sorted keys, compact +separators, UTF-8) prefixed with a purpose tag, so a verdict signature can never +be replayed as an outbox item or a control record and vice versa. + +Only ``cryptography`` beyond the standard library: the kb-gate verifier bundle +vendors this module and pins that dependency by hash. +""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import os +from pathlib import Path +from typing import Any + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, Ed25519PublicKey, +) + +PURPOSES = ("kb-gate-verdict", "kb-outbox-item", "kb-control", "kb-holds", "kb-ack") + + +class SignatureError(ValueError): + """An envelope is malformed, signed for another purpose, or forged.""" + + +def canonical_json(value: Any) -> bytes: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + + +def _message(purpose: str, payload: Any) -> bytes: + if purpose not in PURPOSES: + raise SignatureError(f"unknown signature purpose: {purpose}") + return purpose.encode("ascii") + b"\0" + canonical_json(payload) + + +def public_key_id(public: Ed25519PublicKey) -> str: + raw = public.public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw) + return "ed25519:" + hashlib.sha256(raw).hexdigest()[:16] + + +def public_key_text(public: Ed25519PublicKey) -> str: + """The one-line form committed as ``.github/kb-gate.pub``.""" + raw = public.public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw) + return "ed25519 " + base64.b64encode(raw).decode("ascii") + + +def load_public_key(text: str) -> Ed25519PublicKey: + parts = text.strip().split() + if len(parts) != 2 or parts[0] != "ed25519": + raise SignatureError("public key must be 'ed25519 '") + try: + return Ed25519PublicKey.from_public_bytes(base64.b64decode(parts[1], validate=True)) + except (ValueError, TypeError) as exc: + raise SignatureError("public key is not a valid Ed25519 key") from exc + + +def generate_private_key(path: str | Path) -> Ed25519PrivateKey: + """Create a new private key file readable only by its owner.""" + path = Path(path) + if path.exists(): + raise SignatureError(f"refusing to overwrite an existing key: {path}") + key = Ed25519PrivateKey.generate() + data = key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + path.parent.mkdir(parents=True, exist_ok=True) + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "wb") as handle: + handle.write(data) + return key + + +def load_private_key(path: str | Path) -> Ed25519PrivateKey: + path = Path(path) + mode = path.stat().st_mode & 0o077 + if mode: + raise SignatureError(f"private key must not be readable by group/others: {path}") + key = serialization.load_pem_private_key(path.read_bytes(), password=None) + if not isinstance(key, Ed25519PrivateKey): + raise SignatureError("private key is not Ed25519") + return key + + +def sign(purpose: str, payload: Any, key: Ed25519PrivateKey) -> dict: + signature = key.sign(_message(purpose, payload)) + return { + "purpose": purpose, + "payload": payload, + "key_id": public_key_id(key.public_key()), + "signature": base64.b64encode(signature).decode("ascii"), + } + + +def verify(purpose: str, envelope: Any, public: Ed25519PublicKey) -> Any: + """Return the payload of a valid envelope for ``purpose``; raise otherwise.""" + if not isinstance(envelope, dict) or set(envelope) != {"purpose", "payload", "key_id", "signature"}: + raise SignatureError("envelope must have exactly purpose, payload, key_id and signature") + if envelope["purpose"] != purpose: + raise SignatureError(f"envelope is signed for {envelope['purpose']!r}, not {purpose!r}") + if envelope["key_id"] != public_key_id(public): + raise SignatureError("envelope was signed by a different key") + try: + signature = base64.b64decode(envelope["signature"], validate=True) + public.verify(signature, _message(purpose, envelope["payload"])) + except (InvalidSignature, ValueError, TypeError) as exc: + raise SignatureError("signature does not verify") from exc + return envelope["payload"] diff --git a/test/test_kb_service_core.py b/test/test_kb_service_core.py new file mode 100644 index 000000000..ff73cdab0 --- /dev/null +++ b/test/test_kb_service_core.py @@ -0,0 +1,397 @@ +"""Knowledge service core: lifecycle config, ledger, signing, outbox, kb CLI.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +import pytest +import yaml + +from infermatrix_copilot.adapters.base import AdapterError, RepoAdapter, update_manifest +from infermatrix_copilot.kb_service.config import ( + LifecycleConfigError, general_lifecycle, load_registry, parse_lifecycle, +) +from infermatrix_copilot.kb_service.ledger import Ledger, LeaseError +from infermatrix_copilot.kb_service.outbox import ( + CONTROL_MAX_AGE, Outbox, OutboxError, check_item, verify_holds, +) +from infermatrix_copilot.knowledge_service.signing import ( + SignatureError, generate_private_key, load_private_key, load_public_key, + public_key_text, sign, verify, +) + +ROOT = Path(__file__).resolve().parents[1] + + +class Clock: + def __init__(self, now: float = 1_000_000.0): + self.now = now + + def __call__(self) -> float: + return self.now + + +def _adapter(tmp_path: Path, section: dict, *, auditor: bool = False) -> RepoAdapter: + root = tmp_path / "demo_adapter" + root.mkdir(exist_ok=True) + if auditor: + (root / "audit.py").write_text("def audit_for_knowledge(**kw): return {}\n", encoding="utf-8") + manifest = { + "name": "demo_adapter", + "repo": {"full_name": "org/demo", "path": "/x"}, + "knowledge": {"repo_subdir": "repos/demo"}, + "knowledge_lifecycle": section, + } + (root / "manifest.yaml").write_text(yaml.safe_dump(manifest), encoding="utf-8") + return RepoAdapter(name="demo_adapter", root=root, manifest=manifest) + + +# -- config ---------------------------------------------------------------------- + +def test_shipped_adapters_register_with_repo_neutral_config(): + registry = load_registry(ROOT / "adapters", general=general_lifecycle(enabled=False)) + omni = registry["vllm-omni"] + assert omni.enabled and omni.mode == "shadow" and not omni.auto_merge + assert omni.release.trigger == "github_release" and omni.release.auditor == "release_audit.py" + assert (omni.adapter_dir / omni.release.auditor).is_file() + assert registry["afd-plugin"].enabled is False + assert registry["general"].intake.merged_prs is False + + +@pytest.mark.parametrize("section,message", [ + ({"enabled": True, "mode": "yolo"}, "mode must be one of"), + ({"enabled": True, "mode": "auto_merge", "upstream_visibility": "private", + "calibration_set": "x"}, "private upstream can only run in shadow"), + ({"enabled": True, "mode": "auto_merge"}, "requires a calibration_set"), + ({"enabled": True, "release": {"trigger": "tag_pattern"}}, "tag_pattern is required"), + ({"enabled": True, "release": {"auditor": "missing.py"}}, "auditor does not exist"), + ({"enabled": True, "knowledge_dir": "repos/other"}, "must equal the adapter repo_subdir"), + ({"enabled": True, "surprise": 1}, "unknown keys"), + ({"enabled": "yes"}, "must be true or false"), + ({"enabled": True, "circuit_breaker": {"retire_ratio": 2}}, "out of range"), +]) +def test_invalid_lifecycle_sections_are_refused(tmp_path, section, message): + with pytest.raises(LifecycleConfigError, match=message): + parse_lifecycle(_adapter(tmp_path, section)) + + +def test_private_upstream_never_publishes(tmp_path): + lifecycle = parse_lifecycle(_adapter(tmp_path, {"enabled": True, "upstream_visibility": "private"})) + assert not lifecycle.publishes and not lifecycle.auto_merge + + +def test_lifecycle_section_is_human_only(tmp_path): + adapter = _adapter(tmp_path, {"enabled": True}) + with pytest.raises(AdapterError, match="high-risk"): + update_manifest(adapter, "knowledge_lifecycle", {"enabled": True, "mode": "auto_merge"}, + actor="agent") + + +# -- ledger ---------------------------------------------------------------------- + +def test_ledger_lease_is_exclusive_until_stale(tmp_path): + clock = Clock() + a = Ledger(tmp_path / "kb.db", clock=clock) + b = Ledger(tmp_path / "kb.db", clock=clock) + owner = a.acquire_lease("a", ttl=60) + with pytest.raises(LeaseError): + b.acquire_lease("b", ttl=60) + clock.now += 61 + assert b.acquire_lease("b", ttl=60) == "b" + with pytest.raises(LeaseError): + a.heartbeat(owner) + + +def test_ledger_generations_events_and_repo_isolation(tmp_path): + ledger = Ledger(tmp_path / "kb.db", clock=Clock()) + ledger.ensure_repo("demo", "shadow") + ledger.ensure_repo("other", "shadow") + assert ledger.record_event("demo", "merged_pr", "7", {"n": 7}) is not None + assert ledger.record_event("demo", "merged_pr", "7", {"n": 7}) is None # idempotent + assert ledger.bump_generation("demo", pause=True, reason="breaker") == 2 + assert ledger.repo_state("demo")["paused"] == 1 + assert ledger.repo_state("other")["generation"] == 1 # untouched + assert ledger.resume("demo") == 3 and ledger.repo_state("demo")["paused"] == 0 + cs = ledger.create_changeset("demo", "intake") + assert ledger.changeset(cs)["generation"] == 3 + ledger.record_retirement("demo", "DEMO-1a", "repos/demo/rules.md", "v1") + assert [r["rule_id"] for r in ledger.purge_eligible("demo", "v2")] == ["DEMO-1a"] + assert ledger.purge_eligible("demo", "v1") == [] + + +# -- signing --------------------------------------------------------------------- + +def test_signing_roundtrip_purpose_binding_and_key_permissions(tmp_path): + key = generate_private_key(tmp_path / "k.pem") + assert os.stat(tmp_path / "k.pem").st_mode & 0o077 == 0 + public = load_public_key(public_key_text(key.public_key())) + envelope = sign("kb-gate-verdict", {"pr": 1, "head": "a" * 40}, key) + assert verify("kb-gate-verdict", envelope, public) == {"pr": 1, "head": "a" * 40} + with pytest.raises(SignatureError, match="signed for"): + verify("kb-outbox-item", envelope, public) # no cross-purpose replay + forged = {**envelope, "payload": {"pr": 2, "head": "a" * 40}} + with pytest.raises(SignatureError, match="does not verify"): + verify("kb-gate-verdict", forged, public) + other = generate_private_key(tmp_path / "o.pem") + with pytest.raises(SignatureError, match="different key"): + verify("kb-gate-verdict", sign("kb-gate-verdict", {}, other), public) + with pytest.raises(SignatureError, match="refusing to overwrite"): + generate_private_key(tmp_path / "k.pem") + os.chmod(tmp_path / "k.pem", 0o644) + with pytest.raises(SignatureError, match="readable by group"): + load_private_key(tmp_path / "k.pem") + + +# -- outbox ---------------------------------------------------------------------- + +@pytest.fixture +def world(tmp_path): + clock = Clock() + key = generate_private_key(tmp_path / "service.pem") + ledger = Ledger(tmp_path / "state" / "kb.db", clock=clock) + ledger.ensure_repo("demo", "auto_merge") + outbox = Outbox(tmp_path / "state", key, ledger, clock=clock) + public = load_public_key(public_key_text(key.public_key())) + return clock, ledger, outbox, public, tmp_path / "state" + + +def _read(path: Path): + return json.loads(path.read_text(encoding="utf-8")) + + +def _check(world, item, **kw): + clock, _ledger, _outbox, public, state = world + options = {"repo_publishes": True, "repo_auto_merge": True} + options.update(kw) + return check_item(_read(state / "outbox" / f"{item.id}.json"), _read(state / "outbox" / "control.json"), + public, now=clock.now, **options) + + +def test_publisher_accepts_a_fresh_item(world): + _clock, _ledger, outbox, _public, _state = world + item = outbox.issue("demo", "enqueue", {"pr": 5, "head": "b" * 40}) + outbox.refresh_control() + assert _check(world, item).body["pr"] == 5 + + +def test_generation_bump_voids_everything_issued_before_it(world): + _clock, ledger, outbox, _public, _state = world + enqueue = outbox.issue("demo", "enqueue", {"pr": 5}) + pause = outbox.issue("demo", "pause", {"pr": 5}) + ledger.bump_generation("demo", pause=False, reason="") # rollback / shadow switch + outbox.refresh_control() + with pytest.raises(OutboxError, match="stale generation"): + _check(world, enqueue) + assert _check(world, pause).kind == "pause" # stopping is always allowed + global_old = outbox.issue("demo", "open_pr", {}) + ledger.bump_generation("*", pause=False, reason="") + outbox.refresh_control() + with pytest.raises(OutboxError, match="stale generation"): + _check(world, global_old) + + +@pytest.mark.parametrize("case,message", [ + ("expired", "expired"), + ("stale_control", "control record is stale"), + ("paused", "paused"), + ("private", "private upstream"), + ("shadow", "not in auto_merge"), +]) +def test_publisher_refusals(world, case, message): + clock, ledger, outbox, _public, _state = world + kw = {} + item = outbox.issue("demo", "enqueue", {"pr": 5}) + if case == "paused": + ledger.bump_generation("demo", pause=True, reason="x") + item = outbox.issue("demo", "enqueue", {"pr": 5}) + outbox.refresh_control() + if case == "expired": + clock.now += 31 * 60 + outbox.refresh_control() + if case == "stale_control": + clock.now += CONTROL_MAX_AGE + 1 + if case == "private": + kw["repo_publishes"] = False + if case == "shadow": + kw["repo_auto_merge"] = False + with pytest.raises(OutboxError, match=message): + _check(world, item, **kw) + + +def test_forged_item_is_refused(world): + _clock, _ledger, outbox, _public, state = world + item = outbox.issue("demo", "open_pr", {"title": "ok"}) + outbox.refresh_control() + path = state / "outbox" / f"{item.id}.json" + envelope = _read(path) + envelope["payload"]["body"]["title"] = "evil" + path.write_text(json.dumps(envelope), encoding="utf-8") + with pytest.raises(SignatureError): + _check(world, item) + + +def test_hold_list_names_paused_repos_and_prs_and_expires(world): + clock, ledger, outbox, public, state = world + cs = ledger.create_changeset("demo", "intake") + ledger.update_changeset(cs, status="paused", pr_number=42) + ledger.bump_generation("demo", pause=True, reason="breaker") + outbox.publish_holds(sequence=1, public_repos={"demo"}) + holds = verify_holds(_read(state / "public" / "holds.json"), public, now=clock.now) + assert holds["repos"] == ["demo"] and holds["prs"] == [42] and not holds["global"] + clock.now += 11 * 60 + with pytest.raises(OutboxError, match="stale"): + verify_holds(_read(state / "public" / "holds.json"), public, now=clock.now) + + +def test_acks_must_be_signed_by_the_publisher(world, tmp_path): + _clock, ledger, outbox, _public, state = world + publisher = generate_private_key(tmp_path / "publisher.pem") + publisher_public = load_public_key(public_key_text(publisher.public_key())) + item = outbox.issue("demo", "open_pr", {}) + acks = state / "inbox" / "acks" + acks.mkdir(parents=True) + (acks / f"{item.id}.json").write_text(json.dumps( + sign("kb-ack", {"item_id": item.id, "ok": True, "pr": 9}, publisher)), encoding="utf-8") + impostor = generate_private_key(tmp_path / "impostor.pem") + (acks / "forged.json").write_text(json.dumps( + sign("kb-ack", {"item_id": item.id, "ok": False}, impostor)), encoding="utf-8") + results = outbox.collect_acks(publisher_public) + assert any(r.get("pr") == 9 for r in results) + assert any("invalid" in r for r in results) + assert ledger.outbox_items("demo", "acked")[0]["id"] == item.id + assert (acks / "forged.invalid").exists() + + +# -- CLI ------------------------------------------------------------------------- + +def test_kb_cli_pause_resume_refreshes_signed_control(tmp_path, monkeypatch, capsys): + from infermatrix_copilot.cli.entry import main as cli_main + + key_path = tmp_path / "service.pem" + assert cli_main(["kb", "keygen", "--out", str(key_path)]) == 0 + public = load_public_key(capsys.readouterr().out.strip()) + monkeypatch.setenv("KB_SIGNING_KEY", str(key_path)) + monkeypatch.setenv("KB_STATE_DIR", str(tmp_path / "state")) + monkeypatch.setenv("ADAPTERS_DIR", str(ROOT / "adapters")) + assert cli_main(["kb", "pause", "--repo", "vllm-omni", "--reason", "drill"]) == 0 + control = verify("kb-control", _read(tmp_path / "state" / "outbox" / "control.json"), public) + assert control["repos"]["vllm-omni"]["paused"] is True + holds = verify("kb-holds", _read(tmp_path / "state" / "public" / "holds.json"), public) + assert holds["repos"] == ["vllm-omni"] + assert cli_main(["kb", "resume", "--repo", "vllm-omni"]) == 0 + control = verify("kb-control", _read(tmp_path / "state" / "outbox" / "control.json"), public) + assert control["repos"]["vllm-omni"]["paused"] is False + assert control["repos"]["vllm-omni"]["generation"] == 3 + assert cli_main(["kb", "pause", "--repo", "nope", "--reason", "x"]) == 2 + capsys.readouterr() + assert cli_main(["kb", "status"]) == 0 + status = json.loads(capsys.readouterr().out) + assert {row["repo"] for row in status["repos"]} >= {"vllm-omni", "afd-plugin", "*"} + + +def test_mode_round_trip_voids_items_issued_before_it(world): + _clock, ledger, outbox, _public, _state = world + enqueue = outbox.issue("demo", "enqueue", {"pr": 5}) + assert ledger.ensure_repo("demo", "shadow") is True + assert ledger.ensure_repo("demo", "shadow") is False # no change, no bump + assert ledger.ensure_repo("demo", "auto_merge") is True + outbox.refresh_control() + with pytest.raises(OutboxError, match="stale generation"): + _check(world, enqueue) + + +@pytest.mark.parametrize("mode", ["disabled", "shadow"]) +@pytest.mark.parametrize("kind", ["open_pr", "open_companion_pr", "update_branch", "enqueue", "post_verdict"]) +def test_only_auto_merge_repos_get_non_stop_writes(world, mode, kind): + _clock, ledger, outbox, _public, _state = world + ledger.ensure_repo("demo", mode) + item = outbox.issue("demo", kind, {}) + stop = outbox.issue("demo", "pause", {}) + outbox.refresh_control() + with pytest.raises(OutboxError, match="not in auto_merge"): + _check(world, item) + assert _check(world, stop).kind == "pause" + + +def test_overlapping_refresh_cannot_undo_a_completed_pause(world, tmp_path): + """Two "processes" (threads with their own ledger connections): a refresh + that read the pre-pause state must not publish after the pause does.""" + import threading + + clock, _ledger, outbox, public, state = world + stale_enqueue = outbox.issue("demo", "enqueue", {"pr": 5}) + key = outbox._key + read_done, may_write = threading.Event(), threading.Event() + errors: list[BaseException] = [] + + def refresher(): + try: + slow = Outbox(state, key, Ledger(state / "kb.db", clock=clock), clock=clock) + original = slow._control_payload + + def stalled(): + payload = original() # reads the pre-pause state ... + read_done.set() + may_write.wait(timeout=5) # ... then stalls before writing + return payload + + slow._control_payload = stalled + slow.refresh_control() + except BaseException as exc: # pragma: no cover - surfaced below + errors.append(exc) + + def pauser(): + try: + ledger = Ledger(state / "kb.db", clock=clock) + Outbox(state, key, ledger, clock=clock).transition( + lambda: ledger.bump_generation("demo", pause=True, reason="breaker"), + public_repos={"demo"}) + except BaseException as exc: # pragma: no cover + errors.append(exc) + + first = threading.Thread(target=refresher) + first.start() + assert read_done.wait(timeout=5) + second = threading.Thread(target=pauser) + second.start() + second.join(timeout=0.5) + assert second.is_alive() # the pause waits for the in-flight publication + may_write.set() + first.join(timeout=5) + second.join(timeout=5) + assert not errors, errors + control = verify("kb-control", _read(state / "outbox" / "control.json"), public) + assert control["repos"]["demo"]["paused"] is True + with pytest.raises(OutboxError): + _check(world, stale_enqueue) + + +def test_private_repos_never_appear_in_the_public_hold_list(tmp_path, monkeypatch, capsys): + import shutil + + from infermatrix_copilot.cli.entry import main as cli_main + + adapters = tmp_path / "adapters" + shutil.copytree(ROOT / "adapters", adapters) + secret = adapters / "secret_repo" + secret.mkdir() + (secret / "manifest.yaml").write_text(yaml.safe_dump({ + "name": "secret_repo", + "repo": {"full_name": "org/secret-repo", "path": "/x"}, + "knowledge": {"repo_subdir": "repos/secret-repo"}, + "knowledge_lifecycle": {"enabled": True, "mode": "shadow", "upstream_visibility": "private"}, + }), encoding="utf-8") + key_path = tmp_path / "service.pem" + assert cli_main(["kb", "keygen", "--out", str(key_path)]) == 0 + public = load_public_key(capsys.readouterr().out.strip()) + monkeypatch.setenv("KB_SIGNING_KEY", str(key_path)) + monkeypatch.setenv("KB_STATE_DIR", str(tmp_path / "state")) + monkeypatch.setenv("ADAPTERS_DIR", str(adapters)) + assert cli_main(["kb", "pause", "--repo", "secret-repo", "--reason", "drill"]) == 0 + assert cli_main(["kb", "pause", "--repo", "vllm-omni", "--reason", "drill"]) == 0 + raw = (tmp_path / "state" / "public" / "holds.json").read_text(encoding="utf-8") + assert "secret" not in raw + holds = verify("kb-holds", json.loads(raw), public) + assert holds["repos"] == ["vllm-omni"]