From 13ef8227192591da3661c3597115fdaa95f6540e Mon Sep 17 00:00:00 2001 From: tzhouam Date: Mon, 28 Sep 2026 21:11:19 +0800 Subject: [PATCH] Add the knowledge publisher (kb publish) for the GPU box The only GitHub writer for knowledge PRs. It runs as the repository owner through the box's existing gh login, which never reaches the bot host or any model environment. - publisher: each round reads the service's signed control record over SSH (or a local directory) and does nothing when it is stale. Every outbox item goes through check_item (signature, expiry, generation, pause, and auto_merge in both the control record and the publisher's OWN adapter configuration). The publisher performs the item, records the result locally, then writes a kb-ack signed with the publisher key. Items it cannot verify or must not execute are neither performed nor answered. Every decision is traced to traces/publisher.jsonl. - idempotent: an item already performed only has its ack re-sent; completion records are written atomically, and a corrupt one is set aside for people, never guessed. open_pr reuses an open PR on its branch only when its head is exactly the rebuilt commit (and a draft for companions). It rebuilds the commit deterministically from the signed contents on base_sha in a scratch index, and refuses to overwrite an existing branch with other content. It also refuses any path that is not a governed knowledge page. - actions: open_pr and open_companion_pr (always a draft). post_verdict, enqueue and update_branch first check that the PR is open at the signed head; enqueue marks a paused draft ready, then runs gh pr merge --auto --match-head-commit. pause runs procedure P: dequeue, disable auto-merge, convert to draft (verified), then label kb:hold. close closes the PR. A post_verdict (issued only for an unpaused repository under its current generation) first removes kb:hold and confirms it is gone before commenting, so pause, then resume, then re-gate, then enqueue completes. - a failed pause or close is never acknowledged: it stays in the outbox and is retried every round until it succeeds. Malformed control records and items are traced and skipped; they never stop the daemon. - transport failures (ssh down while listing or delivering an ack) are traced, never fatal: the completion record stays and the ack is re-sent next round without repeating the action. - double gate: without ALLOW_POST=1 a round only records what it would do; pushing a branch also needs ALLOW_PUSH=1. - CLI: kb publish (--remote host:/state | --local dir) [--once]. It needs KB_SERVICE_PUBKEY, KB_PUBLISHER_KEY and KB_PUBLISHER_GIT_AUTHOR, and never opens the service ledger. Co-Authored-By: Claude Opus 5.5 --- doc/architecture/SPEC/kb_service.md | 20 + doc/features/knowledge-service.md | 17 + src/infermatrix_copilot/kb_service/cli.py | 51 +++ .../kb_service/publisher.py | 428 ++++++++++++++++++ test/test_kb_publisher.py | 396 ++++++++++++++++ 5 files changed, 912 insertions(+) create mode 100644 src/infermatrix_copilot/kb_service/publisher.py create mode 100644 test/test_kb_publisher.py 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