diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index c1713e4e..cbd7f08e 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -61,3 +61,23 @@ stdlib + PyYAML + `cryptography`（`kb` extra）+ `.adapters` + `.knowledge_serv 所有页面落定后基线才推进；兜底巡检只做 T1 与 purge。 - 熔断暂停与状态变化在同一加锁事务中发布控制记录与暂停清单。 + +## 2026-09-28 发布器 +`publisher` 在 GPU 盒上以仓库负责人的 gh 登录运行（`kb publish`），是知识 PR 唯一的 GitHub 写入方。 +- 每轮：经 SSH（或本地目录）读取签名控制记录，过期即整轮不动；对每个 outbox 项执行 `check_item`（签名、 + 过期、代际、暂停、控制记录与**发布器自己的适配器配置**同时为 `auto_merge`），执行后先在本地记录结果， + 再写回发布器密钥签名的 `kb-ack`（字段：item_id、kind、changeset_id、ok、pr、head_sha、branch、error）。 +- 幂等：已执行的项只重发 ack（服务收取 ack 后删除该项）；完成记录原子写入，损坏的记录移到 `.corrupt` 并等人处理， + 绝不猜测是否已执行；`open_pr` 只复用 head 恰为重建提交的已打开 PR（companion 还须是 draft）； + 提交由签名内容在 `base_sha` 上用临时索引重建，日期取自该项，重试得到同一 SHA；分支已存在且内容不同则拒绝， + 绝不覆盖。路径必须是受治理的知识页面（`knowledge/{repos/,general}/**.md|yaml`，无 `..`）。 +- 恢复：`post_verdict` 只在未暂停、当前代际下签发，发布它时先移除 `kb:hold` 标签并确认已移除，再发布评论，使评论触发的预检不再因暂停标签失败。 + 失败的 `pause`/`close` 不回执，留在 outbox 中每轮重试直到成功；格式错误的控制记录或项目记入 trace 并跳过。 + SSH 读取或回执失败只记入 trace，不终止进程；完成记录保留，下一轮重发回执而不重复动作。 +- 动作：`open_pr`/`open_companion_pr`（始终 draft）、`post_verdict` 与 `enqueue`/`update_branch`（先核对 PR + 仍打开且 head 未变；`enqueue` 先把 draft 转为 ready，再 `gh pr merge --auto --match-head-commit`）、 + `pause`（程序 P：出队 → 关闭自动合并 → 转 draft，确认已是 draft，再加 `kb:hold`）、`close`。 +- 双重门控：无 `ALLOW_POST=1` 只记录将要执行的动作；推送分支还需 `ALLOW_PUSH=1`。无法验证或不可执行的项 + 既不执行也不回执。每轮与每个决定都写入 `traces/publisher.jsonl`。 +- 测试：`test_kb_publisher.py`（端到端 open_pr → ack → pr_open、dry-run、崩溃后不重复执行、伪造/未配置/过期、 + 越界路径、verdict/入队/暂停/关闭、暂停期间仍执行暂停、SSH 引号与回执、ack 字段）。 diff --git a/doc/features/knowledge-service.md b/doc/features/knowledge-service.md index 5bf87c69..cdce9ffd 100644 --- a/doc/features/knowledge-service.md +++ b/doc/features/knowledge-service.md @@ -89,3 +89,20 @@ python tools/build_kb_gate_bundle.py --lock # 从 PyPI 重新按哈希固定 在 `.github/CODEOWNERS` 中再加入至少一名知识维护者；仓库 admin 开启合并队列（merge commit） 与 ruleset（必需检查 `kb-gate` 限定 GitHub Actions 来源、Code Owner 审阅、新推送需重新批准）。 在此之前，所有触碰 `knowledge/` 的 PR 都会被 `kb-gate` 拒绝（失败即关闭），代码 PR 不受影响。 + +## 发布器（GPU 盒） + +发布器以仓库负责人现有的 `gh` 登录执行服务签名的 outbox 项，写回签名回执。凭据不离开 GPU 盒。 + +```bash +infermatrix-copilot kb keygen --out ~/.infermatrix-copilot/kb-publisher.pem # 发布器密钥；公钥给服务的 KB_PUBLISHER_PUBKEY +export KB_SERVICE_PUBKEY=/path/to/service.pub # 服务公钥（kb keygen 输出的一行） +export KB_PUBLISHER_KEY=~/.infermatrix-copilot/kb-publisher.pem +export KB_PUBLISHER_GIT_AUTHOR='Name ' # 知识 PR 提交的作者 +export KB_PUBLISHER_STATE=/data//kb-publisher # 必须在负责人目录下（gh 包装器按目录选择登录） +infermatrix-copilot kb publish --remote bot-host:/path/to/kb-state --once # 只记录（dry-run） +ALLOW_POST=1 ALLOW_PUSH=1 infermatrix-copilot kb publish --remote bot-host:/path/to/kb-state # 常驻 +``` + +没有 `ALLOW_POST=1` 时只把将要执行的动作写入 `$KB_PUBLISHER_STATE/traces/publisher.jsonl`；推送分支还需要 +`ALLOW_PUSH=1`。控制记录超过 10 分钟未更新（服务可能宕机）时整轮不执行任何动作。 diff --git a/src/infermatrix_copilot/kb_service/cli.py b/src/infermatrix_copilot/kb_service/cli.py index 909a573c..65efd611 100644 --- a/src/infermatrix_copilot/kb_service/cli.py +++ b/src/infermatrix_copilot/kb_service/cli.py @@ -10,6 +10,10 @@ kb serve [--once] the scheduler (holds the single-writer lease) kb activate activate the knowledge snapshot of the repository's main now kb rollback --to SHA point `active` back at an earlier snapshot + kb publish (--remote HOST:/STATE_DIR | --local DIR) [--once] + the publisher (GPU box, owner's gh login): perform + the signed outbox items; writes need ALLOW_POST=1 + (and ALLOW_PUSH=1 to push branches) 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. @@ -20,6 +24,7 @@ import argparse import json import os +import re import sys import time from pathlib import Path @@ -105,6 +110,14 @@ def main(argv: list[str] | None = None) -> int: sub.add_parser("activate") rollback = sub.add_parser("rollback") rollback.add_argument("--to", required=True) + publish = sub.add_parser("publish") + where = publish.add_mutually_exclusive_group(required=True) + where.add_argument("--remote", help="host:/absolute/path of the service state directory (over ssh)") + where.add_argument("--local", help="the service state directory on this machine") + publish.add_argument("--once", action="store_true") + publish.add_argument("--interval", type=float, default=60.0) + publish.add_argument("--publisher-state", + help="default: $KB_PUBLISHER_STATE or ~/.infermatrix-copilot/kb-publisher") args = parser.parse_args(argv) if args.command == "keygen": @@ -114,6 +127,9 @@ def main(argv: list[str] | None = None) -> int: print(public_key_text(key.public_key())) return 0 + if args.command == "publish": + return _publish(args) + state_dir = _state_dir(args.state_dir) ledger = _ledger(state_dir) try: @@ -220,3 +236,38 @@ def _resume(): finally: ledger.close() return 2 + + +def _publish(args) -> int: + """The publisher never opens the service ledger: it only sees the outbox.""" + from ..knowledge_service.signing import load_private_key, load_public_key + from .merge import knowledge_repository + from .publisher import Gh, LocalTransport, Publisher, SshTransport + + for name in ("KB_SERVICE_PUBKEY", "KB_PUBLISHER_KEY", "KB_PUBLISHER_GIT_AUTHOR"): + if not os.environ.get(name): + raise SystemExit(f"{name} is not set") + author = re.fullmatch(r"\s*(.+?)\s*<([^<>\s]+@[^<>\s]+)>\s*", os.environ["KB_PUBLISHER_GIT_AUTHOR"]) + if not author: + raise SystemExit("KB_PUBLISHER_GIT_AUTHOR must look like 'Name '") + state = Path(args.publisher_state or os.environ.get("KB_PUBLISHER_STATE") + or Path.home() / ".infermatrix-copilot" / "kb-publisher").expanduser() + state.mkdir(parents=True, exist_ok=True) + registry = _registry() + publisher = Publisher( + transport=SshTransport.parse(args.remote) if args.remote else LocalTransport(Path(args.local)), + service_public_key=load_public_key(Path(os.environ["KB_SERVICE_PUBKEY"]).read_text(encoding="utf-8")), + publisher_key=load_private_key(os.environ["KB_PUBLISHER_KEY"]), + github=Gh(knowledge_repository(), cwd=state), + state_dir=state, + author=(author.group(1), author.group(2)), + repo_flags={name: (lc.publishes, lc.auto_merge) for name, lc in registry.items() if lc.enabled}, + allow_post=os.environ.get("ALLOW_POST") == "1", + allow_push=os.environ.get("ALLOW_PUSH") == "1", + ) + if args.once: + json.dump(publisher.run_once(), sys.stdout, sort_keys=True) + print() + return 0 + publisher.serve(interval=args.interval) + return 0 diff --git a/src/infermatrix_copilot/kb_service/publisher.py b/src/infermatrix_copilot/kb_service/publisher.py new file mode 100644 index 00000000..9ee35da6 --- /dev/null +++ b/src/infermatrix_copilot/kb_service/publisher.py @@ -0,0 +1,428 @@ +"""The publisher: performs the knowledge service's signed outbox items on GitHub. + +It runs on the GPU box as the repository owner, through that user's existing +``gh`` login, which never leaves the box and never enters a model environment. +The knowledge service holds no GitHub write credential; this is the only place +knowledge PRs are opened, commented on, queued, paused or closed. + +One round (``kb publish --once``; without it, a round per interval): + +1. read the service's signed control record over SSH (or a local directory); + a stale one means the service may be down, and the round does nothing; +2. for every outbox item: verify it with ``outbox.check_item`` (signature, + expiry, generation, pause, the repository's publication mode in BOTH the + control record and the publisher's own adapter configuration), perform it, + record the result locally, then write a signed ack (``kb-ack``) back; +3. an item already performed only has its ack re-sent (the service deletes an + item once it collected the ack), so a crash between the two never performs + an item twice. ``open_pr`` also reuses an open PR on its branch. + +Writes are double-gated like every outward write in the copilot: GitHub is +only written when ``ALLOW_POST=1`` (and, for anything that pushes a branch, +``ALLOW_PUSH=1``); otherwise the round only records what it would do. + +``open_pr`` never trusts a working tree: it rebuilds the change from the +signed file contents on top of the signed ``base_sha`` in a scratch index, +checks every path is a governed knowledge page, pushes the commit to the +item's branch and opens the PR. +""" + +from __future__ import annotations + +import json +import os +import re +import shlex +import subprocess +import tempfile +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable, Mapping, Protocol + +from ..knowledge_service.signing import sign +from .outbox import CONTROL_MAX_AGE, OutboxError, OutboxItem, check_item + +ITEM_ID = re.compile(r"[0-9]+-[0-9a-f]{12}") +GOVERNED_PATH = re.compile(r"knowledge/(?:repos/[A-Za-z0-9._-]+|general)/(?:[A-Za-z0-9._-]+/)*[A-Za-z0-9._-]+\.(?:md|yaml)") +BRANCH = re.compile(r"kb/[A-Za-z0-9._-]+/[A-Za-z0-9._-]+") +HOLD_LABEL = "kb:hold" +PUSHING = frozenset({"open_pr", "open_companion_pr"}) +RETRIED = frozenset({"pause", "close"}) # stops: retried until done, never acked as failed + + +class PublishError(RuntimeError): + """An item could not be performed; it is acked as failed with this reason.""" + + +# -- where the outbox lives ---------------------------------------------------- + +class Transport(Protocol): + def list_items(self) -> list[str]: ... + def read(self, rel: str) -> bytes: ... + def write_ack(self, item_id: str, data: bytes) -> None: ... + + +@dataclass +class LocalTransport: + """The service state directory on this machine (tests, co-located setups).""" + + root: Path + + def list_items(self) -> list[str]: + folder = self.root / "outbox" + return sorted(p.stem for p in folder.glob("*.json") if ITEM_ID.fullmatch(p.stem)) \ + if folder.is_dir() else [] + + def read(self, rel: str) -> bytes: + return (self.root / rel).read_bytes() + + def write_ack(self, item_id: str, data: bytes) -> None: + folder = self.root / "inbox" / "acks" + folder.mkdir(parents=True, exist_ok=True) + tmp = folder / f".{item_id}.tmp" + tmp.write_bytes(data) + tmp.replace(folder / f"{item_id}.json") + + +@dataclass +class SshTransport: + """``host:/path/to/state_dir`` on the bot host, through the box's ssh config.""" + + host: str + root: str + run: Callable[..., subprocess.CompletedProcess] = subprocess.run + + @classmethod + def parse(cls, remote: str) -> "SshTransport": + host, sep, root = remote.partition(":") + if not sep or not host or not root.startswith("/"): + raise SystemExit(f"--remote must be host:/absolute/state_dir, not {remote!r}") + return cls(host, root.rstrip("/")) + + def _ssh(self, command: str, data: bytes | None = None) -> bytes: + proc = self.run(["ssh", "-o", "BatchMode=yes", self.host, command], input=data, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False) + if proc.returncode != 0: + raise OutboxError(f"ssh {self.host}: {proc.stderr.decode(errors='replace')[:300]}") + return proc.stdout + + def list_items(self) -> list[str]: + out = self._ssh(f"ls -1 {shlex.quote(self.root + '/outbox')} 2>/dev/null || true").decode() + return sorted(name[:-5] for name in out.split() if name.endswith(".json") and ITEM_ID.fullmatch(name[:-5])) + + def read(self, rel: str) -> bytes: + return self._ssh(f"cat {shlex.quote(self.root + '/' + rel)}") + + def write_ack(self, item_id: str, data: bytes) -> None: + folder = shlex.quote(self.root + "/inbox/acks") + tmp, final = shlex.quote(f"{self.root}/inbox/acks/.{item_id}.tmp"), \ + shlex.quote(f"{self.root}/inbox/acks/{item_id}.json") + self._ssh(f"mkdir -p {folder} && cat > {tmp} && mv {tmp} {final}", data) + + +# -- GitHub through the owner's gh login ------------------------------------------ + +@dataclass +class Gh: + """``gh`` and ``git`` as the box owner. ``cwd`` must be inside the owner's + directory: the machine's gh wrapper picks the login from it.""" + + repository: str + cwd: Path + run: Callable[..., subprocess.CompletedProcess] = subprocess.run + + def gh(self, *args: str, input: str | None = None, ok_fail: bool = False) -> str: + proc = self.run(["gh", *args], cwd=self.cwd, input=None if input is None else input.encode(), + stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False) + if proc.returncode != 0 and not ok_fail: + raise PublishError(f"gh {' '.join(args[:3])} failed: {proc.stderr.decode(errors='replace')[:300]}") + return proc.stdout.decode(errors="replace") + + def pr(self, number: int) -> dict: + return json.loads(self.gh("pr", "view", str(number), "--repo", self.repository, + "--json", "number,state,isDraft,headRefOid,headRefName,id,labels")) + + +# -- the publisher ------------------------------------------------------------------- + +@dataclass +class Publisher: + transport: Transport + service_public_key: Any + publisher_key: Any + github: Gh + state_dir: Path + repo_flags: Mapping[str, tuple[bool, bool]] # repo -> (publishes, auto_merge), from OUR adapter config + allow_post: bool = False + allow_push: bool = False + author: tuple[str, str] = ("", "") # (name, email) of the commits, the gh login's owner + git_remote: str = "" # default https://github.com/.git + clock: Callable[[], float] = time.time + log: list[dict] = field(default_factory=list) + + # local records ------------------------------------------------------------- + def _done_path(self, item_id: str) -> Path: + return self.state_dir / "done" / f"{item_id}.json" + + def _trace(self, **entry) -> None: + entry = {"at": self.clock(), **entry} + self.log.append(entry) + path = self.state_dir / "traces" / "publisher.jsonl" + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(entry, ensure_ascii=False, sort_keys=True) + "\n") + + def _send_ack(self, ack: dict) -> bool: + """False when the ack could not be delivered; the completion record + stays, so the next round re-sends it without repeating the action.""" + envelope = sign("kb-ack", ack, self.publisher_key) + try: + self.transport.write_ack(ack["item_id"], json.dumps(envelope, ensure_ascii=False).encode()) + return True + except (OutboxError, OSError) as exc: + self._trace(event="ack_undelivered", item=ack["item_id"], error=str(exc)) + return False + + # one round ----------------------------------------------------------------- + def run_once(self) -> dict: + summary = {"performed": 0, "failed": 0, "resent": 0, "undelivered": 0, "dry_run": 0, "skipped": 0} + try: + control = json.loads(self.transport.read("outbox/control.json")) + # a quick look before verifying (check_item verifies it for real) + control_issued = float(control["payload"]["issued_at"]) + except (OutboxError, OSError, ValueError, KeyError, TypeError) as exc: + self._trace(event="no_control", error=str(exc)) + return summary + if self.clock() - control_issued > CONTROL_MAX_AGE: + self._trace(event="stale_control", issued_at=control_issued) + return summary # the service may be down: act on nothing + try: + item_ids = self.transport.list_items() + except (OutboxError, OSError) as exc: + self._trace(event="outbox_unreachable", error=str(exc)) + return summary + for item_id in item_ids: + done = self._done_path(item_id) + if done.is_file(): + try: + record = json.loads(done.read_text(encoding="utf-8")) + except ValueError: + # never guess whether it was performed: set it aside for people + done.replace(done.with_suffix(".corrupt")) + self._trace(event="corrupt_record", item=item_id) + summary["skipped"] += 1 + continue + summary["resent" if self._send_ack(record) else "undelivered"] += 1 + continue + if done.with_suffix(".corrupt").exists(): + summary["skipped"] += 1 # waits for a person to decide (see the trace) + continue + summary[self._handle(item_id, control)] += 1 + return summary + + def _record_done(self, item_id: str, ack: dict) -> None: + """Atomically: a crash leaves either no record or a complete one.""" + path = self._done_path(item_id) + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(prefix=f".{item_id}.", dir=path.parent) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(json.dumps(ack, ensure_ascii=False, sort_keys=True)) + handle.flush() + os.fsync(handle.fileno()) + os.replace(tmp, path) + except BaseException: + Path(tmp).unlink(missing_ok=True) + raise + + def _handle(self, item_id: str, control: Any) -> str: + try: + envelope = json.loads(self.transport.read(f"outbox/{item_id}.json")) + payload = envelope.get("payload") if isinstance(envelope, dict) else None + claimed = str(payload.get("repo", "")) if isinstance(payload, dict) else "" + publishes, auto_merge = self.repo_flags.get(claimed, (False, False)) + item = check_item(envelope, control, self.service_public_key, now=self.clock(), + repo_publishes=publishes, repo_auto_merge=auto_merge) + except (OutboxError, OSError, ValueError, KeyError, TypeError) as exc: + # not performed and not acked: an unverifiable item is never + # answered, and a stale one is superseded by the service itself + self._trace(event="refused", item=item_id, error=str(exc)) + return "skipped" + if not self.allow_post or (item.kind in PUSHING and not self.allow_push): + self._trace(event="dry_run", item=item.id, kind=item.kind, repo=item.repo, + body={k: v for k, v in item.body.items() if k not in ("files", "comment", "body")}) + return "dry_run" + ack = {"item_id": item.id, "kind": item.kind, "changeset_id": str(item.body.get("changeset_id", "")), + "ok": True, "pr": None, "head_sha": "", "branch": "", "error": ""} + try: + ack.update(getattr(self, f"_do_{item.kind}")(item)) + except (PublishError, OSError, ValueError, KeyError, TypeError) as exc: + ack.update(ok=False, error=str(exc)[:500]) + if not ack["ok"] and item.kind in RETRIED: + # a stop that did not happen must not be answered as done: the item + # stays in the outbox and is retried every round until it succeeds + self._trace(event="retry", item=item.id, kind=item.kind, repo=item.repo, error=ack["error"]) + return "failed" + self._record_done(item.id, ack) + self._trace(event="performed" if ack["ok"] else "failed", item=item.id, kind=item.kind, + repo=item.repo, ack=ack) + self._send_ack(ack) + return "performed" if ack["ok"] else "failed" + + # the actions --------------------------------------------------------------- + def _checked_head(self, item: OutboxItem) -> dict: + pr = self.github.pr(int(item.body["pr"])) + if pr.get("state") != "OPEN": + raise PublishError(f"PR #{item.body['pr']} is {pr.get('state')}") + if pr.get("headRefOid") != item.body["head_sha"]: + raise PublishError(f"PR #{item.body['pr']} head moved to {str(pr.get('headRefOid'))[:12]}") + return pr + + def _do_open_pr(self, item: OutboxItem, *, draft: bool = False) -> dict: + body = item.body + branch = str(body["branch"]) + if not BRANCH.fullmatch(branch): + raise PublishError(f"refusing branch name {branch!r}") + files: dict[str, str] = dict(body.get("files") or {}) + deleted = [str(p) for p in body.get("deleted") or []] + for path in [*files, *deleted]: + if not GOVERNED_PATH.fullmatch(path) or ".." in path.split("/"): + raise PublishError(f"refusing to write outside governed knowledge pages: {path}") + commit = self._build_commit(str(body["base_sha"]), files, deleted, str(body["title"]), + when=item.issued_at) + existing = json.loads(self.github.gh("pr", "list", "--repo", self.github.repository, "--head", branch, + "--state", "open", "--json", "number,headRefOid,isDraft")) + if existing: # performed before a crash: reuse it only if it is exactly this change + pr = existing[0] + if len(existing) != 1 or pr["headRefOid"] != commit: + raise PublishError(f"an open PR on {branch} does not carry the signed change") + if draft and not pr.get("isDraft"): + raise PublishError(f"the PR on {branch} must be a draft") + return {"pr": int(pr["number"]), "head_sha": commit, "branch": branch} + remote = self.git_remote or f"https://github.com/{self.github.repository}.git" + listed = self._git("ls-remote", remote, f"refs/heads/{branch}", auth=True).split() + if not listed: # expect the branch to be absent: never overwrite anything + self._git("push", "--quiet", remote, f"{commit}:refs/heads/{branch}", + f"--force-with-lease=refs/heads/{branch}:", auth=True) + elif listed[0] != commit: + raise PublishError(f"branch {branch} already exists with other content") + args = ["pr", "create", "--repo", self.github.repository, "--base", "main", "--head", branch, + "--title", str(body["title"]), "--body-file", "-"] + self.github.gh(*args, *(["--draft"] if draft else []), input=str(body.get("body") or "")) + opened = json.loads(self.github.gh("pr", "list", "--repo", self.github.repository, "--head", branch, + "--state", "open", "--json", "number,headRefOid")) + if len(opened) != 1 or opened[0]["headRefOid"] != commit: + raise PublishError(f"the PR for {branch} could not be confirmed after creation") + return {"pr": int(opened[0]["number"]), "head_sha": commit, "branch": branch} + + def _do_open_companion_pr(self, item: OutboxItem) -> dict: + return self._do_open_pr(item, draft=True) # companions are always drafts for people + + def _do_post_verdict(self, item: OutboxItem) -> dict: + """A verdict is only issued for an unpaused repository, under its current + generation, so posting one also lifts an earlier pause's kb:hold label + (first, so the precheck the comment triggers no longer sees it).""" + self._checked_head(item) + number = str(item.body["pr"]) + self.github.gh("pr", "edit", number, "--repo", self.github.repository, "--remove-label", HOLD_LABEL, + ok_fail=True) # fails when the label is absent; what matters is checked next + if HOLD_LABEL in {str(label.get("name")) for label in self.github.pr(int(number)).get("labels") or []}: + raise PublishError(f"PR #{number} still carries {HOLD_LABEL}; the gate would reject it") + self.github.gh("pr", "comment", number, "--repo", self.github.repository, + "--body-file", "-", input=str(item.body["comment"])) + return {"pr": int(item.body["pr"]), "head_sha": item.body["head_sha"]} + + def _do_enqueue(self, item: OutboxItem) -> dict: + pr = self._checked_head(item) + number = str(item.body["pr"]) + if pr.get("isDraft"): # a resumed PR was turned into a draft by its pause + self.github.gh("pr", "ready", number, "--repo", self.github.repository) + self.github.gh("pr", "merge", number, "--repo", self.github.repository, "--merge", "--auto", + "--match-head-commit", str(item.body["head_sha"])) + return {"pr": int(number), "head_sha": item.body["head_sha"]} + + def _do_pause(self, item: OutboxItem) -> dict: + """Procedure P: out of the merge queue, then draft (the guarantee), then + the kb:hold label (a record only).""" + number = str(item.body["pr"]) + pr = self.github.pr(int(number)) + if pr.get("state") != "OPEN": + return {"pr": int(number)} # merged or closed already: nothing to stop + self.github.gh("api", "graphql", "-f", "query=mutation($id:ID!){dequeuePullRequest(input:{id:$id})" + "{clientMutationId}}", "-f", f"id={pr['id']}", ok_fail=True) + self.github.gh("pr", "merge", number, "--repo", self.github.repository, "--disable-auto", ok_fail=True) + if not pr.get("isDraft"): + self.github.gh("pr", "ready", number, "--repo", self.github.repository, "--undo") + if not self.github.pr(int(number)).get("isDraft"): + raise PublishError(f"PR #{number} is still not a draft") + self.github.gh("pr", "edit", number, "--repo", self.github.repository, "--add-label", HOLD_LABEL, + ok_fail=True) + return {"pr": int(number)} + + def _do_update_branch(self, item: OutboxItem) -> dict: + self._checked_head(item) + self.github.gh("api", "-X", "PUT", f"repos/{self.github.repository}/pulls/{item.body['pr']}/update-branch", + "-f", f"expected_head_sha={item.body['head_sha']}") + return {"pr": int(item.body["pr"])} + + def _do_close(self, item: OutboxItem) -> dict: + number = str(item.body["pr"]) + if self.github.pr(int(number)).get("state") == "OPEN": + self.github.gh("pr", "close", number, "--repo", self.github.repository, + "--comment", str(item.body.get("reason") or "closed by the knowledge service")) + return {"pr": int(number)} + + # git ------------------------------------------------------------------------ + @property + def clone(self) -> Path: + return self.state_dir / "clones" / self.github.repository.replace("/", "__") + + def _git(self, *args: str, env: Mapping[str, str] | None = None, input: bytes | None = None, + auth: bool = False) -> str: + prefix = ["-c", "credential.helper=", "-c", "credential.helper=!gh auth git-credential"] if auth else [] + proc = subprocess.run(["git", "-C", str(self.clone), *prefix, *args], input=input, + env={**os.environ, **(env or {})}, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, check=False, cwd=self.github.cwd) + if proc.returncode != 0: + raise PublishError(f"git {args[0]} failed: {proc.stderr.decode(errors='replace')[:300]}") + return proc.stdout.decode().strip() + + def _build_commit(self, base_sha: str, files: Mapping[str, str], deleted: list[str], title: str, + *, when: float) -> str: + """The change as ONE commit on ``base_sha``, built in a scratch index. + Deterministic (dates from the item), so a retry rebuilds the same SHA.""" + remote = self.git_remote or f"https://github.com/{self.github.repository}.git" + if not (self.clone / ".git").exists() and not (self.clone / "HEAD").exists(): + self.clone.parent.mkdir(parents=True, exist_ok=True) + proc = subprocess.run(["git", "clone", "--quiet", "--bare", remote, str(self.clone)], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, cwd=self.github.cwd) + if proc.returncode != 0: + raise PublishError(f"git clone failed: {proc.stderr.decode(errors='replace')[:300]}") + self._git("fetch", "--quiet", remote, "+refs/heads/main:refs/remotes/origin/main") + if self._git("cat-file", "-t", base_sha) != "commit": + raise PublishError(f"base {base_sha[:12]} is not a commit") + with tempfile.TemporaryDirectory() as scratch: + env = {"GIT_INDEX_FILE": str(Path(scratch) / "index")} + self._git("read-tree", base_sha, env=env) + for path, text in sorted(files.items()): + oid = self._git("hash-object", "-w", "--stdin", input=text.encode("utf-8")) + self._git("update-index", "--add", "--cacheinfo", f"100644,{oid},{path}", env=env) + for path in deleted: + self._git("update-index", "--force-remove", path, env=env) + tree = self._git("write-tree", env=env) + name, email = self.author + if not name or not email: + raise PublishError("no commit author configured (KB_PUBLISHER_GIT_AUTHOR='Name ')") + date = f"{int(when)} +0000" + return self._git("commit-tree", tree, "-p", base_sha, "-m", title, env={ + "GIT_AUTHOR_NAME": name, "GIT_AUTHOR_EMAIL": email, "GIT_AUTHOR_DATE": date, + "GIT_COMMITTER_NAME": name, "GIT_COMMITTER_EMAIL": email, "GIT_COMMITTER_DATE": date}) + + def serve(self, *, interval: float, stop=None) -> None: + import threading + + stop = stop or threading.Event() + while not stop.is_set(): + self._trace(event="round", **self.run_once()) + stop.wait(interval) diff --git a/test/test_kb_publisher.py b/test/test_kb_publisher.py new file mode 100644 index 00000000..bdac2014 --- /dev/null +++ b/test/test_kb_publisher.py @@ -0,0 +1,396 @@ +"""The GPU-box publisher: performs signed outbox items through gh, acks back, offline.""" + +from __future__ import annotations + +import json +import subprocess +from pathlib import Path + +import pytest + +from infermatrix_copilot.kb_service import merge +from infermatrix_copilot.kb_service.publisher import Gh, LocalTransport, Publisher, SshTransport +from infermatrix_copilot.knowledge_service.signing import generate_private_key, sign +from test_kb_flow import _flow_runtime, _open_pr +from test_kb_intake_gate import _git as _git_raw + +REPO = "JiusiServe/InferMatrixCopilot" + + +def _git(path: Path, *args: str) -> str: + return _git_raw(path, *args).strip() + + +class FakeGh: + """Just enough of `gh` for the publisher; PR heads come from the git remote.""" + + def __init__(self, origin: Path): + self.origin = origin + self.prs: dict[int, dict] = {} + self.calls: list[list[str]] = [] + self.comments: list[tuple[int, str]] = [] + self.fail: set[str] = set() # "remove-label" / "undo": that call fails + + def _by_branch(self, branch: str) -> list[dict]: + return [pr for pr in self.prs.values() if pr["headRefName"] == branch and pr["state"] == "OPEN"] + + def __call__(self, argv, *, cwd=None, input=None, stdout=None, stderr=None, check=False): + assert argv[0] == "gh" + args = argv[1:] + self.calls.append(args) + out = "" + if args[:2] == ["pr", "list"]: + branch = args[args.index("--head") + 1] + out = json.dumps([{"number": p["number"], "headRefOid": p["headRefOid"], "isDraft": p["isDraft"]} + for p in self._by_branch(branch)]) + elif args[:2] == ["pr", "create"]: + branch = args[args.index("--head") + 1] + head = _git(self.origin, "rev-parse", f"refs/heads/{branch}") + number = 42 + len(self.prs) + self.prs[number] = {"number": number, "state": "OPEN", "isDraft": "--draft" in args, + "headRefOid": head, "headRefName": branch, "id": f"PR_{number}", + "labels": [], "queued": False, "body": (input or b"").decode()} + elif args[:2] == ["pr", "view"]: + pr = self.prs[int(args[2])] + out = json.dumps({**{k: pr[k] for k in ("number", "state", "isDraft", "headRefOid", "headRefName", "id")}, + "labels": [{"name": name} for name in pr["labels"]]}) + elif args[:2] == ["pr", "comment"]: + self.comments.append((int(args[2]), input.decode())) + elif args[:2] == ["pr", "ready"]: + if "--undo" in args and "undo" in self.fail: + self.fail.discard("undo") # fails once + return subprocess.CompletedProcess(argv, 1, b"", b"HTTP 502") + self.prs[int(args[2])]["isDraft"] = "--undo" in args + elif args[:2] == ["pr", "merge"]: + pr = self.prs[int(args[2])] + if "--disable-auto" in args: + pr["queued"] = False + else: + assert args[args.index("--match-head-commit") + 1] == pr["headRefOid"] + pr["queued"] = True + elif args[:2] == ["pr", "edit"]: + labels = self.prs[int(args[2])]["labels"] + if "--remove-label" in args and "remove-label" in self.fail: + return subprocess.CompletedProcess(argv, 1, b"", b"HTTP 502") + if "--add-label" in args: + labels.append(args[args.index("--add-label") + 1]) + elif args[args.index("--remove-label") + 1] in labels: + labels.remove(args[args.index("--remove-label") + 1]) + elif args[:2] == ["pr", "close"]: + self.prs[int(args[2])]["state"] = "CLOSED" + elif args[:2] == ["api", "graphql"]: + for pr in self.prs.values(): + if f"id={pr['id']}" in args: + pr["queued"] = False + elif args[:3] == ["api", "-X", "PUT"]: + pass + else: + raise AssertionError(args) + return subprocess.CompletedProcess(argv, 0, out.encode(), b"") + + def created(self) -> int: + return sum(1 for call in self.calls if call[:2] == ["pr", "create"]) + + +def _setup(tmp_path, *, allow=True, flags=None): + rt, lifecycle = _flow_runtime(tmp_path) + changeset_id, publisher_key = _open_pr(tmp_path, rt, lifecycle) + rt.outbox.refresh_control() # `kb serve` does this every tick + origin = tmp_path / "origin" + gh = FakeGh(origin) + pub = Publisher( + transport=LocalTransport(tmp_path / "state"), + service_public_key=rt.outbox._key.public_key(), + publisher_key=publisher_key, + github=Gh(REPO, cwd=tmp_path, run=gh), + state_dir=tmp_path / "publisher", + repo_flags=flags if flags is not None else {lifecycle.repo: (True, True)}, + allow_post=allow, allow_push=allow, + author=("tzhouam", "tzhouam@connect.ust.hk"), + git_remote=str(origin), + clock=rt.clock, + ) + return rt, lifecycle, changeset_id, pub, gh + + +def _collect(rt): + merge.apply_acks(rt, rt.outbox.collect_acks(rt.publisher_public_key)) + + +def _issue(rt, lifecycle, changeset_id, kind, **body): + return rt.outbox.issue(lifecycle.repo, kind, {"changeset_id": changeset_id, **body}) + + +def test_open_pr_rebuilds_the_signed_change_on_its_base_and_acks_the_pr(tmp_path): + rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path) + assert pub.run_once()["performed"] == 1 + _collect(rt) + changeset = rt.ledger.changeset(changeset_id) + assert changeset["status"] == "pr_open" and changeset["pr_number"] == 42 + branch = f"kb/{lifecycle.repo}/{changeset_id}" + head = _git(tmp_path / "origin", "rev-parse", f"refs/heads/{branch}") + assert changeset["head_sha"] == head and changeset["branch"] == branch + data = rt.load_changeset_files(changeset_id) + assert _git(tmp_path / "origin", "rev-parse", f"{head}^") == data["base_sha"] + changed = _git(tmp_path / "origin", "diff", "--name-only", f"{head}^", head).split() + assert changed == sorted(f"knowledge/{rel}" for rel in data["files"]) + for rel, text in data["files"].items(): + assert _git_raw(tmp_path / "origin", "show", f"{head}:knowledge/{rel}") == text + assert _git(tmp_path / "origin", "log", "-1", "--format=%an <%ae>", head) == "tzhouam " + assert gh.prs[42]["isDraft"] is False and "changeset" in gh.prs[42]["body"] + + +def test_without_allow_post_the_round_only_records(tmp_path): + rt, _lifecycle, changeset_id, pub, gh = _setup(tmp_path, allow=False) + assert pub.run_once()["dry_run"] == 1 + assert gh.calls == [] and not (tmp_path / "state" / "inbox" / "acks").exists() + trace = (tmp_path / "publisher" / "traces" / "publisher.jsonl").read_text() + assert '"event": "dry_run"' in trace and '"files"' not in trace + pub.allow_post = True # posting alone never pushes a branch + assert pub.run_once()["dry_run"] == 1 and gh.calls == [] + assert rt.ledger.changeset(changeset_id)["status"] == "pr_requested" + + +def test_a_crash_after_performing_never_performs_twice(tmp_path): + rt, _lifecycle, changeset_id, pub, gh = _setup(tmp_path) + pub.run_once() + for ack in (tmp_path / "state" / "inbox" / "acks").glob("*.json"): + ack.unlink() # the ack never reached the service + assert pub.run_once()["resent"] == 1 and gh.created() == 1 + for done in (tmp_path / "publisher" / "done").glob("*.json"): + done.unlink() # crashed before even recording it + for ack in (tmp_path / "state" / "inbox" / "acks").glob("*.json"): + ack.unlink() + assert pub.run_once()["performed"] == 1 and gh.created() == 1 # reuses the open PR + _collect(rt) + assert rt.ledger.changeset(changeset_id)["pr_number"] == 42 + + +def test_unverifiable_unconfigured_or_stale_items_are_never_performed(tmp_path): + rt, lifecycle, _changeset_id, pub, gh = _setup(tmp_path, flags={}) + assert pub.run_once()["skipped"] == 1 and gh.calls == [] # our config says: not auto_merge + pub.repo_flags = {lifecycle.repo: (True, True)} + pub.service_public_key = generate_private_key(tmp_path / "other.pem").public_key() + assert pub.run_once()["skipped"] == 1 and gh.calls == [] # not signed by the service + pub.service_public_key = rt.outbox._key.public_key() + now = rt.clock() + pub.clock = lambda: now + 11 * 60 # control record went stale + assert pub.run_once() == {"performed": 0, "failed": 0, "resent": 0, "undelivered": 0, + "dry_run": 0, "skipped": 0} + assert gh.calls == [] + + +def test_open_pr_refuses_paths_outside_governed_knowledge_pages(tmp_path): + rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path) + for path in list((tmp_path / "state" / "outbox").glob("[0-9]*.json")): + path.unlink() + data = rt.load_changeset_files(changeset_id) + for bad in ("src/x.py", "knowledge/tools/x.md", "knowledge/repos/demo/../../.github/x.md"): + _issue(rt, lifecycle, f"{changeset_id}-{len(bad)}", "open_pr", base_sha=data["base_sha"], + branch=f"kb/{lifecycle.repo}/x{len(bad)}", files={bad: "x\n"}, deleted=[], title="t", body="b") + assert pub.run_once()["failed"] == 3 and gh.created() == 0 + acks = [json.loads(p.read_text())["payload"] for p in (tmp_path / "state" / "inbox" / "acks").glob("*.json")] + assert all(not a["ok"] and "outside governed knowledge pages" in a["error"] for a in acks) + + +def test_post_verdict_enqueue_pause_and_close(tmp_path): + rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path) + pub.run_once() + head = gh.prs[42]["headRefOid"] + _issue(rt, lifecycle, changeset_id, "post_verdict", pr=42, head_sha="f" * 40, comment="stale") + _issue(rt, lifecycle, changeset_id, "post_verdict", pr=42, head_sha=head, comment="") + summary = pub.run_once() + assert summary["performed"] == 1 and summary["failed"] == 1 + assert gh.comments == [(42, "")] + gh.prs[42]["isDraft"] = True # paused earlier, then resumed + rt.outbox.refresh_control() + _issue(rt, lifecycle, changeset_id, "enqueue", pr=42, head_sha=head) + assert pub.run_once()["performed"] == 1 + assert gh.prs[42]["isDraft"] is False and gh.prs[42]["queued"] is True + _issue(rt, lifecycle, changeset_id, "pause", pr=42, reason="drill") + assert pub.run_once()["performed"] == 1 + assert gh.prs[42]["queued"] is False and gh.prs[42]["isDraft"] is True and "kb:hold" in gh.prs[42]["labels"] + _issue(rt, lifecycle, changeset_id, "close", pr=42, reason="superseded") + _issue(rt, lifecycle, changeset_id, "pause", pr=42, reason="after close") + assert pub.run_once()["performed"] == 2 and gh.prs[42]["state"] == "CLOSED" + + +def test_pause_is_performed_even_while_the_repository_is_paused(tmp_path): + rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path) + pub.run_once() + _collect(rt) # the PR is open, so the pause applies to it + rt.outbox.transition(lambda: rt.ledger.bump_generation(lifecycle.repo, pause=True, reason="drill"), + public_repos={lifecycle.repo}) + merge.pause_open_prs(rt.ledger, rt.outbox, lifecycle.repo, "drill") + _collect(rt) + assert pub.run_once()["performed"] == 1 and gh.prs[42]["isDraft"] is True + _collect(rt) + assert rt.ledger.changeset(changeset_id)["status"] == "paused" + + +def test_pause_resume_regate_and_enqueue_round_trip(tmp_path): + """pause (draft + kb:hold) -> kb resume -> fresh verdict lifts the label -> + the gate can pass -> enqueue readies the draft and queues it.""" + rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path) + pub.run_once() + _collect(rt) + head = gh.prs[42]["headRefOid"] + rt.github.prs[42] = {"number": 42, "state": "open", "merged": False, "head": {"sha": head}} + rt.outbox.transition(lambda: rt.ledger.bump_generation(lifecycle.repo, pause=True, reason="drill"), + public_repos={lifecycle.repo}) + merge.pause_open_prs(rt.ledger, rt.outbox, lifecycle.repo, "drill") + pub.run_once() + _collect(rt) + assert gh.prs[42]["isDraft"] and "kb:hold" in gh.prs[42]["labels"] + assert rt.ledger.changeset(changeset_id)["status"] == "paused" + + def _resume(): + rt.ledger.resume(lifecycle.repo) + merge.resume_paused_prs(rt.ledger, lifecycle.repo) + rt.outbox.transition(_resume, public_repos={lifecycle.repo}) + assert merge.advance(rt, lifecycle) == [f"verdict issued {changeset_id}"] + pub.run_once() + _collect(rt) + assert "kb:hold" not in gh.prs[42]["labels"] and gh.comments[-1][0] == 42 + rt.github.statuses[head] = "success" # the precheck the comment triggered + assert merge.advance(rt, lifecycle) == [f"enqueue issued {changeset_id}"] + pub.run_once() + _collect(rt) + assert gh.prs[42]["isDraft"] is False and gh.prs[42]["queued"] is True + assert rt.ledger.changeset(changeset_id)["status"] == "queued" + + +def test_a_failed_pause_is_retried_not_acknowledged(tmp_path): + rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path) + pub.run_once() + _collect(rt) + gh.prs[42]["queued"] = True + gh.fail.add("undo") + rt.outbox.transition(lambda: rt.ledger.bump_generation(lifecycle.repo, pause=True, reason="drill"), + public_repos={lifecycle.repo}) + merge.pause_open_prs(rt.ledger, rt.outbox, lifecycle.repo, "drill") + assert pub.run_once()["failed"] == 1 + assert not list((tmp_path / "state" / "inbox" / "acks").glob("*.json")) # nothing answered + assert pub.run_once()["performed"] == 1 and gh.prs[42]["isDraft"] is True + _collect(rt) + assert rt.ledger.changeset(changeset_id)["status"] == "paused" + + +@pytest.mark.parametrize("control", [{"payload": {"issued_at": "invalid"}}, [], {"payload": []}, "x"]) +def test_malformed_control_records_and_items_never_stop_the_publisher(tmp_path, control): + rt, _lifecycle, _changeset_id, pub, gh = _setup(tmp_path) + good = (tmp_path / "state" / "outbox" / "control.json").read_text() + (tmp_path / "state" / "outbox" / "control.json").write_text(json.dumps(control)) + assert pub.run_once()["performed"] == 0 and gh.calls == [] + (tmp_path / "state" / "outbox" / "control.json").write_text(good) + (tmp_path / "state" / "outbox" / "1-bbbbbbbbbbbb.json").write_text(json.dumps(control)) + summary = pub.run_once() + assert summary["skipped"] == 1 and summary["performed"] == 1 + + +def test_a_verdict_is_not_posted_while_the_hold_label_cannot_be_removed(tmp_path): + rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path) + pub.run_once() + head = gh.prs[42]["headRefOid"] + gh.prs[42]["labels"].append("kb:hold") + gh.fail.add("remove-label") + _issue(rt, lifecycle, changeset_id, "post_verdict", pr=42, head_sha=head, comment="verdict") + assert pub.run_once()["failed"] == 1 and gh.comments == [] + + +def test_transport_failures_keep_the_daemon_alive_and_resend_later(tmp_path): + rt, _lifecycle, changeset_id, pub, gh = _setup(tmp_path) + real = pub.transport + + class Flaky: + down = True + + def list_items(self): + return real.list_items() + + def read(self, rel): + return real.read(rel) + + def write_ack(self, item_id, data): + if self.down: + raise OSError("ssh: connection reset") + real.write_ack(item_id, data) + + pub.transport = Flaky() + assert pub.run_once()["performed"] == 1 # performed; its ack could not be delivered + assert not list((tmp_path / "state" / "inbox" / "acks").glob("*.json")) + pub.transport.down = False + assert pub.run_once()["resent"] == 1 and gh.created() == 1 + _collect(rt) + assert rt.ledger.changeset(changeset_id)["status"] == "pr_open" + + class Unreachable: + def read(self, rel): + return real.read(rel) + + def list_items(self): + raise OSError("ssh: no route to host") + + pub.transport = Unreachable() + assert pub.run_once()["performed"] == 0 # traced, not raised + assert "outbox_unreachable" in (tmp_path / "publisher" / "traces" / "publisher.jsonl").read_text() + + +def test_an_open_pr_with_other_content_is_never_adopted(tmp_path): + rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path) + branch = f"kb/{lifecycle.repo}/{changeset_id}" + gh.prs[7] = {"number": 7, "state": "OPEN", "isDraft": False, "headRefOid": "e" * 40, + "headRefName": branch, "id": "PR_7", "labels": [], "queued": False, "body": ""} + assert pub.run_once()["failed"] == 1 and gh.created() == 0 + (ack,) = [json.loads(p.read_text())["payload"] for p in (tmp_path / "state" / "inbox" / "acks").glob("*.json")] + assert not ack["ok"] and "does not carry the signed change" in ack["error"] + + +def test_a_corrupt_completion_record_is_set_aside_not_guessed(tmp_path): + rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path) + (item_id,) = pub.transport.list_items() + done = tmp_path / "publisher" / "done" / f"{item_id}.json" + done.parent.mkdir(parents=True) + done.write_text('{"item_id": "tru') # a write interrupted by a crash + assert pub.run_once()["skipped"] == 1 and gh.calls == [] + assert done.with_suffix(".corrupt").exists() and not done.exists() + assert pub.run_once()["skipped"] == 1 and gh.calls == [] # still waits for a person + assert "corrupt_record" in (tmp_path / "publisher" / "traces" / "publisher.jsonl").read_text() + assert not list((tmp_path / "publisher" / "done").glob(".*")) # atomic writes leave no temp files + + +def test_ssh_transport_quotes_paths_and_pipes_acks(): + seen = [] + + def run(argv, input=None, **kw): + seen.append((argv, input)) + out = b"1-aaaaaaaaaaaa.json\ncontrol.json\nbad name.json\n" if "ls -1" in argv[-1] else b"{}" + return subprocess.CompletedProcess(argv, 0, out, b"") + + transport = SshTransport("bot", "/srv/kb state", run=run) + assert transport.list_items() == ["1-aaaaaaaaaaaa"] + transport.read("outbox/control.json") + transport.write_ack("1-aaaaaaaaaaaa", b"ACK") + assert seen[0][0][:4] == ["ssh", "-o", "BatchMode=yes", "bot"] + assert "'/srv/kb state/outbox/control.json'" in seen[1][0][-1] + assert seen[2][1] == b"ACK" and "mv '/srv/kb state/inbox/acks/.1-aaaaaaaaaaaa.tmp'" in seen[2][0][-1] + with pytest.raises(SystemExit): + SshTransport.parse("bot:relative/path") + + +def test_kb_publish_needs_both_keys(monkeypatch, tmp_path): + from infermatrix_copilot.kb_service.cli import main + + monkeypatch.delenv("KB_SERVICE_PUBKEY", raising=False) + with pytest.raises(SystemExit, match="KB_SERVICE_PUBKEY"): + main(["publish", "--local", str(tmp_path), "--once"]) + + +def test_acks_carry_the_documented_fields(tmp_path): + rt, _lifecycle, _changeset_id, pub, _gh = _setup(tmp_path) + pub.run_once() + (ack_file,) = (tmp_path / "state" / "inbox" / "acks").glob("*.json") + envelope = json.loads(ack_file.read_text()) + assert envelope["purpose"] == "kb-ack" + assert set(envelope["payload"]) == {"item_id", "kind", "changeset_id", "ok", "pr", "head_sha", "branch", "error"} + assert sign # the envelope format is the shared signing module's