From ee3d7cf5a35987cf3d1f0f5161a0255d37f93cb9 Mon Sep 17 00:00:00 2001 From: Zhang Kezhen <120164259+ApolloZhangOnGithub@users.noreply.github.com> Date: Sun, 17 May 2026 17:04:19 +0800 Subject: [PATCH] =?UTF-8?q?cnb:=20ownership=20routing=20L1=20=E2=80=94=20a?= =?UTF-8?q?ssignee/label/path=20priority=20+=20per-file=20CI=20(#87)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the binary substring scan with explicit evidence + confidence, per the design comment on #87. ## Issue routing — 3-tier priority chain 1. **assignee** (high) — GitHub `assignees[].login` matched against known sessions. User picked this person; respect it before anything else. 2. **label** (high) — `proj:` / `area:` matched against the tail segment of an ownership path. `proj:fetch_bilibili` → owner of `lib/fetch_bilibili/`. 3. **path** (medium) — regex over body/title catches `lib/foo.py` etc., then `find_owner()` per path. Single match → that owner. Multiple distinct owners → escalate to fallback (no broadcast). 4. **substring** (low) — legacy behavior preserved for ownership patterns mentioned without a full filename. Unmatched / multi-owner cases route to a single fallback recipient (`lead` if a `lead` session exists, otherwise broadcast `all`) with `ambiguous:` or `no_match` evidence — not multi-ping. The orphaned-owner re-route to `all` is preserved. ## CI routing — per-file, not broadcast `_scan_ci` now calls `gh run view --json files` to get the failed run's changed files, runs them through `find_owner`, and notifies only the owners of those files. When `gh run view` fails / files can't be determined, ONE fallback message goes to `lead`/`all` — no longer pings every distinct owner. ## Audit log New `routing_log` table records every decision (kind / ref / recipient / evidence / confidence). `board own audit [--limit N]` shows recent entries for misroute debrief. Migration 010 creates the table; schema.sql mirrors it. ## Message bodies surface the evidence Every routed message now ends with: ``` matched-via: assigned:bezos # or label:proj:foo, path:lib/x.py, ... confidence: high # high|medium|low|fallback ``` so the receiver can challenge a misroute without re-running the scan. ## Tests 22 new tests in `tests/test_ownership_routing.py` covering the matrix from the design comment (assignee wins, label wins, path single/multi, substring fallback, no-match fallback, lead-preferred fallback, scan end-to-end with audit log, CI per-file routing, CI timeout fallback, audit subcommand happy/empty/limit/invalid). One existing test (`test_scan_ci_failure`) updated for the new per-file behavior. Full suite green: 1776 tests pass. Closes #87 (L1 scope; confidence-from-feedback and ML routing remain L2/L3). Co-Authored-By: Claude Opus 4.7 (1M context) --- CHANGELOG.md | 6 + VERSION | 2 +- lib/board_own.py | 285 ++++++++++++++++--- migrations/011_routing_log.sql | 15 + package.json | 2 +- pyproject.toml | 2 +- schema.sql | 13 + tests/test_ownership.py | 24 +- tests/test_ownership_routing.py | 478 ++++++++++++++++++++++++++++++++ 9 files changed, 777 insertions(+), 50 deletions(-) create mode 100644 migrations/011_routing_log.sql create mode 100644 tests/test_ownership_routing.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 3e3ae7c..99b8ee6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## 0.5.93-dev (unreleased) + +### Features + +- **Ownership routing L1 (#87)** — Rewrote `board scan` issue and CI routing with explicit evidence and confidence. Issue routing follows a 3-tier priority chain: GitHub `assignees` (high confidence) → `proj:*` / `area:*` labels (high) → file-path references in title/body (medium) → existing substring fallback (low). Multi-owner matches and unmatched issues route to a fallback recipient (`lead` if it exists, else `all`) with `ambiguous:` or `no_match` evidence rather than broadcasting. CI failures now route per-file via `gh run view --json files` — when the failed run's changed files map to specific owners, only those owners are notified; when files can't be determined (timeout / no data), one fallback message goes to `lead`/`all` instead of the previous spammy broadcast to every owner. Every routing decision is recorded in a new `routing_log` table; `board own audit [--limit N]` shows recent decisions for misroute debrief. Notification message bodies always include `matched-via:` and `confidence:` lines so receivers can spot misrouting. Closes #87 (L1 scope; confidence-from-feedback and ML-based routing remain L2/L3). + ## 0.5.78-dev (unreleased) ### Features diff --git a/VERSION b/VERSION index c892f02..50de1d1 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.5.80-dev +0.5.93-dev diff --git a/lib/board_own.py b/lib/board_own.py index ee7330d..b5846b1 100644 --- a/lib/board_own.py +++ b/lib/board_own.py @@ -2,6 +2,7 @@ import json import os +import re import subprocess import tomllib from dataclasses import dataclass @@ -13,6 +14,14 @@ DEFAULT_ORPHAN_HOURS = 24 +# Issue routing (#87 L1): patterns to extract file paths from issue bodies. +# Catches things like `lib/board_own.py`, `docs/getting-started.md`, +# `tests/test_x.py`. Restricted to common cnb source extensions to avoid +# matching random words. +_PATH_PATTERN_RE = re.compile( + r"\b((?:lib|bin|tests|docs|migrations|cnb)/[\w/.-]+\.(?:py|md|toml|json|sql|sh|yml|yaml))\b" +) + def cmd_own(db: BoardDB, identity: str, args: list[str]) -> None: validate_identity(db, identity) @@ -35,8 +44,10 @@ def cmd_own(db: BoardDB, identity: str, args: list[str]) -> None: _own_orphans(db, rest) elif subcmd == "map": _own_map(db) + elif subcmd == "audit": + _own_audit(db, rest) else: - print("Usage: board --as own {claim|list|disown|transfer|transfer-all|offboard|orphans|map}") + print("Usage: board --as own {claim|list|disown|transfer|transfer-all|offboard|orphans|map|audit}") raise SystemExit(1) @@ -633,11 +644,120 @@ def cmd_scan(db: BoardDB, identity: str, args: list[str]) -> None: print("OK scan 完成: 无新事项") +@dataclass(frozen=True) +class RouteDecision: + """One routing decision: who to notify, why, and how strong the signal is.""" + + recipient: str + evidence: str + confidence: str # "high" | "medium" | "low" | "fallback" + + +def _known_sessions(db: BoardDB) -> set[str]: + return {str(row[0]) for row in db.query("SELECT name FROM sessions WHERE name NOT IN ('all', 'system')")} + + +def _fallback_recipient(db: BoardDB) -> str: + """Where unmatched / ambiguous routing goes. Prefer `lead`, else broadcast `all`.""" + row = db.query_one("SELECT name FROM sessions WHERE name='lead'") + return "lead" if row else "all" + + +def _route_issue(db: BoardDB, issue: dict, ownership_rows: list[tuple[str, str]]) -> RouteDecision: + """Decide where one GitHub issue should go. Priority: assignee > label > path > fallback. + + The first source that yields ≥1 candidate wins. Multi-owner matches in + the path tier escalate to fallback (no broadcast-style pings). + """ + title = str(issue.get("title", "")) + body = str(issue.get("body") or "") + assignees = issue.get("assignees") or [] + labels = issue.get("labels") or [] + known = _known_sessions(db) + by_path: dict[str, str] = {p: s for s, p in ownership_rows} + + # 1. Assignee — highest confidence (user picked this person on GitHub). + for a in assignees: + login = (a.get("login") if isinstance(a, dict) else str(a)).lower() + if login in known: + return RouteDecision(login, f"assigned:{login}", "high") + + # 2. Label `proj:` / `area:` — explicit project tag points at owner. + for label in labels: + name = (label.get("name") if isinstance(label, dict) else str(label)).lower() + if not (name.startswith("proj:") or name.startswith("area:")): + continue + tag = name.split(":", 1)[1] + # match label tag against owned-path tail: pattern "lib/foo/" matches label "foo" + for pattern, session in [(p, by_path[p]) for p in by_path]: + parts = [seg for seg in pattern.strip("/").split("/") if seg] + if tag and parts and tag in parts: + return RouteDecision(session, f"label:{name}", "high") + + # 3. Path references in body — regex over common cnb extensions. + referenced_paths = _PATH_PATTERN_RE.findall(f"{title}\n{body}") + path_owners: dict[str, str] = {} # owner → evidence path + for fpath in referenced_paths: + owner = find_owner(db, fpath) + if owner and owner not in path_owners: + path_owners[owner] = fpath + if len(path_owners) == 1: + owner, fpath = next(iter(path_owners.items())) + return RouteDecision(owner, f"path:{fpath}", "medium") + if len(path_owners) > 1: + return RouteDecision( + _fallback_recipient(db), + f"ambiguous:{','.join(sorted(path_owners))}", + "fallback", + ) + + # 4. Substring fallback (preserve legacy behavior for issue bodies that + # mention an ownership pattern without a full file path, e.g. "lib/"). + text = f"{title}\n{body}".lower() + substring_owners: list[tuple[str, str]] = [] # (session, pattern) + for session, pattern in ownership_rows: + if pattern.lower() in text: + substring_owners.append((session, pattern)) + if len(substring_owners) == 1: + session, pattern = substring_owners[0] + return RouteDecision(session, f"substring:{pattern}", "low") + if len(substring_owners) > 1: + owners = sorted({s for s, _ in substring_owners}) + return RouteDecision( + _fallback_recipient(db), + f"ambiguous:{','.join(owners)}", + "fallback", + ) + + # 5. No match. + return RouteDecision(_fallback_recipient(db), "no_match", "fallback") + + +def _record_routing(db: BoardDB, kind: str, ref: str, decision: RouteDecision) -> None: + """Append one row to routing_log. Silent on schema-missing — older DBs are tolerated.""" + try: + db.execute( + "INSERT INTO routing_log(kind, ref, recipient, evidence, confidence) VALUES (?, ?, ?, ?, ?)", + (kind, ref, decision.recipient, decision.evidence, decision.confidence), + ) + except Exception: + pass # tolerate pre-migration DBs + + +def _format_issue_routing_body(number: int, title: str, decision: RouteDecision, *, orphan_note: str = "") -> str: + """Render the message body. Always includes matched-via + confidence so the + receiver can spot misrouting without re-running the scan.""" + header = f"[ISSUE #{number}] {title}" + if orphan_note: + header = f"{header} — {orphan_note}" + return f"{header}\nmatched-via: {decision.evidence}\nconfidence: {decision.confidence}" + + def _scan_issues(db: BoardDB, project_root: Path) -> int: - """Check open GitHub issues, notify owners of relevant ones.""" + """Check open GitHub issues, notify owners using L1 routing (#87).""" try: r = subprocess.run( - ["gh", "issue", "list", "--state", "open", "--json", "number,title,labels,body"], + ["gh", "issue", "list", "--state", "open", "--json", "number,title,labels,body,assignees"], cwd=str(project_root), capture_output=True, text=True, @@ -650,47 +770,66 @@ def _scan_issues(db: BoardDB, project_root: Path) -> int: except (subprocess.TimeoutExpired, FileNotFoundError, json.JSONDecodeError): return 0 - ownership_rows = db.query("SELECT session, path_pattern FROM ownership") + ownership_rows = [(s, p) for s, p in db.query("SELECT session, path_pattern FROM ownership")] if not ownership_rows: return 0 routed = 0 for issue in issues: - title = issue.get("title", "") - body = issue.get("body", "") or "" number = issue.get("number", 0) - text = f"{title} {body}".lower() - - for session, pattern in ownership_rows: - if pattern.lower() in text: - is_orphan = _is_orphaned_owner(db, session) - recipient = "all" if is_orphan else session - already = db.scalar( - "SELECT COUNT(*) FROM messages WHERE body LIKE ? AND recipient=?", - (f"%[ISSUE #{number}]%", recipient), - ) - if not already: - if is_orphan: - body = ( - f"[ISSUE #{number}] {title} — 原 owner {session} 可能 orphaned，相关 ownership: {pattern}" - ) - else: - body = f"[ISSUE #{number}] {title} — 可能与你负责的 {pattern} 相关" - db.post_message( - "system", - recipient, - body, - deliver=True, - ) - routed += 1 + title = issue.get("title", "") + decision = _route_issue(db, issue, ownership_rows) + + # Orphan owner re-route: preserve legacy behavior of sending to "all" + # with the original owner mentioned in the body. + orphan_note = "" + if decision.recipient in _known_sessions(db) and _is_orphaned_owner(db, decision.recipient): + orphan_note = f"原 owner {decision.recipient} 可能 orphaned, 路由证据: {decision.evidence}" + decision = RouteDecision("all", f"orphan:{decision.recipient}", "fallback") + + already = db.scalar( + "SELECT COUNT(*) FROM messages WHERE body LIKE ? AND recipient=?", + (f"%[ISSUE #{number}]%", decision.recipient), + ) + if already: + continue + + body = _format_issue_routing_body(number, title, decision, orphan_note=orphan_note) + db.post_message("system", decision.recipient, body, deliver=True) + _record_routing(db, "issue", f"#{number}", decision) + routed += 1 return routed +def _failed_run_files(project_root: Path, run_id: int | str) -> list[str]: + """Get the list of changed files for a failed run via `gh run view`. + + Returns [] on any error / timeout — callers fall back to lead routing. + """ + try: + r = subprocess.run( + ["gh", "run", "view", str(run_id), "--json", "files"], + cwd=str(project_root), + capture_output=True, + text=True, + timeout=15, + ) + if r.returncode != 0: + return [] + data = json.loads(r.stdout) + files = data.get("files") if isinstance(data, dict) else None + if not isinstance(files, list): + return [] + return [str(f.get("path") or f) for f in files if f] + except (subprocess.TimeoutExpired, FileNotFoundError, json.JSONDecodeError): + return [] + + def _scan_ci(db: BoardDB, project_root: Path) -> int: - """Check CI status of current branch, notify owners of failing files.""" + """Route CI failures by per-file ownership (#87 L1), not broadcast.""" try: r = subprocess.run( - ["gh", "run", "list", "--limit", "1", "--json", "status,conclusion,headBranch"], + ["gh", "run", "list", "--limit", "1", "--json", "status,conclusion,headBranch,databaseId"], cwd=str(project_root), capture_output=True, text=True, @@ -704,22 +843,78 @@ def _scan_ci(db: BoardDB, project_root: Path) -> int: return 0 branch = runs[0].get("headBranch", "unknown") + run_id = runs[0].get("databaseId", 0) except (subprocess.TimeoutExpired, FileNotFoundError, json.JSONDecodeError): return 0 - owners = db.query("SELECT DISTINCT session FROM ownership") + files = _failed_run_files(project_root, run_id) if run_id else [] + owners: dict[str, list[str]] = {} # owner → [matching files] + for fpath in files: + owner = find_owner(db, fpath) + if owner: + owners.setdefault(owner, []).append(fpath) + routed = 0 - for (session,) in owners: - already = db.scalar( - "SELECT COUNT(*) FROM messages WHERE body LIKE ? AND recipient=?", - (f"%[CI FAIL]%{branch}%", session), - ) - if not already: - db.post_message( - "system", - session, - f"[CI FAIL] {branch} 分支 CI 失败，请检查你负责的模块", - deliver=True, + if owners: + # Route to each matched owner with the specific files as evidence. + for owner, matched_files in owners.items(): + ref = f"{branch}:{run_id}" + already = db.scalar( + "SELECT COUNT(*) FROM messages WHERE body LIKE ? AND recipient=?", + (f"%[CI FAIL]%{ref}%", owner), + ) + if already: + continue + evidence = f"files:{','.join(matched_files[:3])}" + confidence = "high" + body = ( + f"[CI FAIL] {ref} 失败，涉及你负责的文件: {', '.join(matched_files[:3])}\n" + f"matched-via: {evidence}\n" + f"confidence: {confidence}" ) + db.post_message("system", owner, body, deliver=True) + _record_routing(db, "ci", ref, RouteDecision(owner, evidence, confidence)) routed += 1 - return routed + return routed + + # Fallback: couldn't determine files → route ONE message to lead (or all), + # not broadcast to every owner. Better than the previous spammy behavior. + fallback = _fallback_recipient(db) + ref = f"{branch}:{run_id or 'unknown'}" + already = db.scalar( + "SELECT COUNT(*) FROM messages WHERE body LIKE ? AND recipient=?", + (f"%[CI FAIL]%{ref}%", fallback), + ) + if already: + return 0 + body = f"[CI FAIL] {ref} 分支 CI 失败，无法定位 owner — 请人工分派\nmatched-via: no_files\nconfidence: fallback" + db.post_message("system", fallback, body, deliver=True) + _record_routing(db, "ci", ref, RouteDecision(fallback, "no_files", "fallback")) + return 1 + + +def _own_audit(db: BoardDB, args: list[str]) -> None: + """Show recent routing decisions from routing_log.""" + flags, _ = parse_flags(args, value_flags={"limit": ["--limit", "-n"]}) + try: + limit = int(flags["limit"]) if "limit" in flags else 20 + except ValueError: + print("ERROR: --limit 必须是整数") + raise SystemExit(1) + + try: + rows = db.query( + "SELECT ts, kind, ref, recipient, evidence, confidence FROM routing_log ORDER BY id DESC LIMIT ?", + (limit,), + ) + except Exception: + print("无 routing_log（schema 未迁移?）") + return + + if not rows: + print("routing_log 为空") + return + + print(f"Routing audit (最近 {len(rows)} 条):") + for ts, kind, ref, recipient, evidence, confidence in rows: + print(f" [{ts}] {kind} {ref} → {recipient} [{confidence}] {evidence}") diff --git a/migrations/011_routing_log.sql b/migrations/011_routing_log.sql new file mode 100644 index 0000000..76574ba --- /dev/null +++ b/migrations/011_routing_log.sql @@ -0,0 +1,15 @@ +-- Audit trail for board scan routing decisions (#87 L1). +-- Each row records one issue/CI notification dispatched by _scan_issues +-- or _scan_ci, including the evidence that justified the recipient choice. +-- Used by `board own audit` for misroute debrief. +CREATE TABLE IF NOT EXISTS routing_log( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts TEXT NOT NULL DEFAULT (strftime('%Y-%m-%d %H:%M:%S','now','localtime')), + kind TEXT NOT NULL, -- 'issue' | 'ci' + ref TEXT NOT NULL, -- '#42' | 'branch:job' + recipient TEXT NOT NULL, -- session name or 'lead' or 'all' + evidence TEXT NOT NULL, -- 'assigned:bezos' | 'label:proj:foo' | 'path:lib/x.py' | 'fallback:no_match' + confidence TEXT NOT NULL -- 'high' | 'medium' | 'low' | 'fallback' +); +CREATE INDEX IF NOT EXISTS idx_routing_log_ref ON routing_log(ref); +CREATE INDEX IF NOT EXISTS idx_routing_log_recipient ON routing_log(recipient); diff --git a/package.json b/package.json index 9519177..99de1c5 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "claude-nb", - "version": "0.5.78-dev", + "version": "0.5.93-dev", "description": "Multi-agent coordination framework for Claude Code sessions", "engines": { "node": ">=18" diff --git a/pyproject.toml b/pyproject.toml index 6d9a159..c34c842 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "claude-nb" -version = "0.5.78.dev0" +version = "0.5.93.dev0" description = "Multi-agent coordination framework for Claude Code sessions" requires-python = ">=3.11" license = "MIT" diff --git a/schema.sql b/schema.sql index 690d2b4..4521679 100644 --- a/schema.sql +++ b/schema.sql @@ -199,3 +199,16 @@ CREATE TABLE IF NOT EXISTS mail( CREATE INDEX IF NOT EXISTS idx_mail_thread ON mail(thread_id); CREATE INDEX IF NOT EXISTS idx_mail_sender ON mail(sender); CREATE INDEX IF NOT EXISTS idx_mail_ts ON mail(ts); + +-- Audit trail for board scan routing decisions (#87 L1). +CREATE TABLE IF NOT EXISTS routing_log( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts TEXT NOT NULL DEFAULT (strftime('%Y-%m-%d %H:%M:%S','now','localtime')), + kind TEXT NOT NULL, + ref TEXT NOT NULL, + recipient TEXT NOT NULL, + evidence TEXT NOT NULL, + confidence TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_routing_log_ref ON routing_log(ref); +CREATE INDEX IF NOT EXISTS idx_routing_log_recipient ON routing_log(recipient); diff --git a/tests/test_ownership.py b/tests/test_ownership.py index 1426b57..ae68e3f 100644 --- a/tests/test_ownership.py +++ b/tests/test_ownership.py @@ -511,6 +511,8 @@ def side_effect(cmd, **kwargs): @patch("lib.board_own.subprocess.run") def test_scan_ci_failure(self, mock_run, db, capsys): + """L1 (#87): CI failure routes per-file. When the failed run touches + a file owned by alice, alice is notified — not broadcast to everyone.""" env = MagicMock() env.project_root = Path("/tmp/fake") db.env = env @@ -521,10 +523,25 @@ def test_scan_ci_failure(self, mock_run, db, capsys): def side_effect(cmd, **kwargs): if "issue" in cmd: return MagicMock(returncode=0, stdout="[]") - if "run" in cmd: + if "run" in cmd and "list" in cmd: + return MagicMock( + returncode=0, + stdout=json.dumps( + [ + { + "status": "completed", + "conclusion": "failure", + "headBranch": "feature-x", + "databaseId": 12345, + } + ] + ), + ) + if "run" in cmd and "view" in cmd: + # gh run view --json files → files touched by this run return MagicMock( returncode=0, - stdout=json.dumps([{"status": "completed", "conclusion": "failure", "headBranch": "feature-x"}]), + stdout=json.dumps({"files": [{"path": "lib/board_own.py"}, {"path": "lib/other.py"}]}), ) return MagicMock(returncode=1, stdout="") @@ -536,3 +553,6 @@ def side_effect(cmd, **kwargs): msgs = db.query("SELECT body FROM messages WHERE recipient='alice' AND body LIKE '%CI FAIL%'") assert len(msgs) == 1 + # L1 evidence: matched-via path is in the body so receiver can sanity-check. + assert "matched-via: files:" in msgs[0][0] + assert "lib/board_own.py" in msgs[0][0] diff --git a/tests/test_ownership_routing.py b/tests/test_ownership_routing.py new file mode 100644 index 0000000..eb6fc75 --- /dev/null +++ b/tests/test_ownership_routing.py @@ -0,0 +1,478 @@ +"""Tests for #87 L1 ownership routing: 3-tier priority + per-file CI + audit log. + +See `lib/board_own.py::_route_issue` and `lib/board_own._scan_ci`. The matrix +covers the acceptance criteria defined in the design comment on #87: + +- Issue routing: assignee > label > path > substring > no-match fallback +- Multi-owner matches escalate to lead/all fallback (no broadcast pings) +- Orphaned owner re-routes to "all" with the original owner mentioned +- CI routing: per-file owner notification, fallback to lead/all when files + cannot be determined +- routing_log audit table records every decision +""" + +import json +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from lib.board_own import ( + RouteDecision, + _route_issue, + cmd_own, + cmd_scan, +) + +# --------------------------------------------------------------------------- +# Tier 1: assignee +# --------------------------------------------------------------------------- + + +class TestRouteByAssignee: + def test_assignee_matches_known_session(self, db): + cmd_own(db, "alice", ["claim", "docs/"]) + ownership_rows = [("alice", "docs/")] + issue = { + "number": 1, + "title": "anything", + "body": "", + "labels": [], + "assignees": [{"login": "bob"}], + } + decision = _route_issue(db, issue, ownership_rows) + assert decision.recipient == "bob" + assert decision.evidence == "assigned:bob" + assert decision.confidence == "high" + + def test_assignee_wins_over_label_and_path(self, db): + cmd_own(db, "alice", ["claim", "lib/foo/"]) + ownership_rows = [("alice", "lib/foo/")] + issue = { + "number": 2, + "title": "touches lib/foo/bar.py and label proj:foo", + "body": "see lib/foo/bar.py", + "labels": [{"name": "proj:foo"}], + "assignees": [{"login": "charlie"}], + } + decision = _route_issue(db, issue, ownership_rows) + assert decision.recipient == "charlie" + assert decision.evidence == "assigned:charlie" + + def test_unknown_assignee_falls_through(self, db): + cmd_own(db, "alice", ["claim", "docs/"]) + ownership_rows = [("alice", "docs/")] + issue = { + "number": 3, + "title": "docs", + "body": "in docs/", + "labels": [], + "assignees": [{"login": "stranger"}], # not in sessions + } + decision = _route_issue(db, issue, ownership_rows) + # falls through to substring tier (docs/ matches alice's pattern) + assert decision.recipient == "alice" + + +# --------------------------------------------------------------------------- +# Tier 2: label +# --------------------------------------------------------------------------- + + +class TestRouteByLabel: + def test_proj_label_maps_to_owner_via_path_tail(self, db): + cmd_own(db, "alice", ["claim", "lib/fetch_bilibili/"]) + ownership_rows = [("alice", "lib/fetch_bilibili/")] + issue = { + "number": 4, + "title": "anything", + "body": "", + "labels": [{"name": "proj:fetch_bilibili"}], + "assignees": [], + } + decision = _route_issue(db, issue, ownership_rows) + assert decision.recipient == "alice" + assert decision.evidence == "label:proj:fetch_bilibili" + assert decision.confidence == "high" + + def test_area_label_also_supported(self, db): + cmd_own(db, "bob", ["claim", "lib/board_own/"]) + ownership_rows = [("bob", "lib/board_own/")] + issue = { + "number": 5, + "title": "anything", + "body": "", + "labels": [{"name": "area:board_own"}], + "assignees": [], + } + decision = _route_issue(db, issue, ownership_rows) + assert decision.recipient == "bob" + assert decision.evidence == "label:area:board_own" + + def test_unknown_label_tag_falls_through(self, db): + cmd_own(db, "alice", ["claim", "lib/foo/"]) + ownership_rows = [("alice", "lib/foo/")] + issue = { + "number": 6, + "title": "lib/foo/bar.py issue", + "body": "", + "labels": [{"name": "proj:nonexistent"}], + "assignees": [], + } + decision = _route_issue(db, issue, ownership_rows) + # falls through to path tier + assert decision.recipient == "alice" + assert decision.evidence.startswith("path:") + + def test_label_wins_over_path(self, db): + cmd_own(db, "alice", ["claim", "lib/foo/"]) + cmd_own(db, "bob", ["claim", "lib/bar/"]) + ownership_rows = [("alice", "lib/foo/"), ("bob", "lib/bar/")] + issue = { + "number": 7, + "title": "touches lib/bar/x.py but tagged foo", + "body": "lib/bar/x.py", + "labels": [{"name": "proj:foo"}], + "assignees": [], + } + decision = _route_issue(db, issue, ownership_rows) + assert decision.recipient == "alice" + assert decision.evidence == "label:proj:foo" + + +# --------------------------------------------------------------------------- +# Tier 3: path references +# --------------------------------------------------------------------------- + + +class TestRouteByPath: + def test_single_path_match(self, db): + cmd_own(db, "alice", ["claim", "lib/"]) + ownership_rows = [("alice", "lib/")] + issue = { + "number": 8, + "title": "bug in lib/board_view.py", + "body": "see lib/board_view.py:123", + "labels": [], + "assignees": [], + } + decision = _route_issue(db, issue, ownership_rows) + assert decision.recipient == "alice" + assert decision.evidence == "path:lib/board_view.py" + assert decision.confidence == "medium" + + def test_multi_owner_path_match_escalates_to_fallback(self, db): + cmd_own(db, "alice", ["claim", "lib/"]) + cmd_own(db, "bob", ["claim", "tests/"]) + ownership_rows = [("alice", "lib/"), ("bob", "tests/")] + issue = { + "number": 9, + "title": "lib/x.py + tests/test_x.py both broken", + "body": "lib/x.py at line 5\ntests/test_x.py:9", + "labels": [], + "assignees": [], + } + decision = _route_issue(db, issue, ownership_rows) + # No lead session in default fixture → fallback is "all" + assert decision.recipient == "all" + assert decision.evidence.startswith("ambiguous:") + assert "alice" in decision.evidence and "bob" in decision.evidence + assert decision.confidence == "fallback" + + +# --------------------------------------------------------------------------- +# Substring fallback (legacy behavior preserved) +# --------------------------------------------------------------------------- + + +class TestRouteBySubstring: + def test_substring_match_low_confidence(self, db): + cmd_own(db, "alice", ["claim", "lib/"]) + ownership_rows = [("alice", "lib/")] + issue = { + "number": 10, + "title": "lib/ improvements", + "body": "various lib/ cleanups", + "labels": [], + "assignees": [], + } + decision = _route_issue(db, issue, ownership_rows) + assert decision.recipient == "alice" + assert decision.evidence == "substring:lib/" + assert decision.confidence == "low" + + def test_ambiguous_substring_escalates(self, db): + cmd_own(db, "alice", ["claim", "lib/"]) + cmd_own(db, "bob", ["claim", "tests/"]) + ownership_rows = [("alice", "lib/"), ("bob", "tests/")] + issue = { + "number": 11, + "title": "lib/ and tests/ mention", + "body": "", + "labels": [], + "assignees": [], + } + decision = _route_issue(db, issue, ownership_rows) + assert decision.recipient == "all" + assert decision.evidence.startswith("ambiguous:") + + +# --------------------------------------------------------------------------- +# No match fallback +# --------------------------------------------------------------------------- + + +class TestRouteFallback: + def test_no_match_routes_to_fallback(self, db): + cmd_own(db, "alice", ["claim", "lib/zoo/"]) + ownership_rows = [("alice", "lib/zoo/")] + issue = { + "number": 12, + "title": "totally unrelated", + "body": "no path no label no assignee", + "labels": [], + "assignees": [], + } + decision = _route_issue(db, issue, ownership_rows) + # No lead session in fixture → fallback to "all" + assert decision.recipient == "all" + assert decision.evidence == "no_match" + assert decision.confidence == "fallback" + + def test_lead_session_preferred_as_fallback(self, db): + # Add a lead session so fallback prefers it over "all" + db.execute("INSERT OR IGNORE INTO sessions(name) VALUES ('lead')") + cmd_own(db, "alice", ["claim", "lib/zoo/"]) + ownership_rows = [("alice", "lib/zoo/")] + issue = { + "number": 13, + "title": "no match", + "body": "", + "labels": [], + "assignees": [], + } + decision = _route_issue(db, issue, ownership_rows) + assert decision.recipient == "lead" + + +# --------------------------------------------------------------------------- +# Scan end-to-end: routing decision becomes a posted message + audit row +# --------------------------------------------------------------------------- + + +class TestScanIntegration: + @patch("lib.board_own.subprocess.run") + def test_scan_writes_evidence_to_message_body(self, mock_run, db, capsys): + env = MagicMock() + env.project_root = Path("/tmp/fake") + db.env = env + cmd_own(db, "alice", ["claim", "lib/"]) + capsys.readouterr() + + issue = { + "number": 99, + "title": "lib/board_view.py bug", + "body": "lib/board_view.py crashes", + "labels": [], + "assignees": [], + } + mock_run.side_effect = lambda cmd, **_: ( + MagicMock(returncode=0, stdout=json.dumps([issue])) + if "issue" in cmd + else MagicMock(returncode=0, stdout="[]") + ) + + cmd_scan(db, "alice", []) + capsys.readouterr() + + row = db.query_one("SELECT body FROM messages WHERE recipient='alice' AND body LIKE '%ISSUE #99%'") + assert row is not None + body = row[0] + assert "matched-via: path:lib/board_view.py" in body + assert "confidence: medium" in body + + @patch("lib.board_own.subprocess.run") + def test_scan_records_routing_log(self, mock_run, db, capsys): + env = MagicMock() + env.project_root = Path("/tmp/fake") + db.env = env + cmd_own(db, "alice", ["claim", "lib/"]) + capsys.readouterr() + + issue = { + "number": 100, + "title": "lib/x.py issue", + "body": "", + "labels": [], + "assignees": [], + } + mock_run.side_effect = lambda cmd, **_: ( + MagicMock(returncode=0, stdout=json.dumps([issue])) + if "issue" in cmd + else MagicMock(returncode=0, stdout="[]") + ) + + cmd_scan(db, "alice", []) + capsys.readouterr() + + row = db.query_one("SELECT kind, ref, recipient, evidence, confidence FROM routing_log WHERE ref='#100'") + assert row is not None + kind, _ref, recipient, evidence, confidence = row + assert kind == "issue" + assert recipient == "alice" + assert evidence == "path:lib/x.py" + assert confidence == "medium" + + +# --------------------------------------------------------------------------- +# CI routing (per-file) +# --------------------------------------------------------------------------- + + +class TestCiPerFileRouting: + @patch("lib.board_own.subprocess.run") + def test_files_route_to_their_owner_only(self, mock_run, db, capsys): + env = MagicMock() + env.project_root = Path("/tmp/fake") + db.env = env + cmd_own(db, "alice", ["claim", "lib/"]) + cmd_own(db, "bob", ["claim", "tests/"]) + capsys.readouterr() + + def side_effect(cmd, **_): + if "issue" in cmd: + return MagicMock(returncode=0, stdout="[]") + if "run" in cmd and "list" in cmd: + return MagicMock( + returncode=0, + stdout=json.dumps( + [{"status": "completed", "conclusion": "failure", "headBranch": "f1", "databaseId": 1}] + ), + ) + if "run" in cmd and "view" in cmd: + # only lib/ files touched — alice should get the ping, bob should not + return MagicMock( + returncode=0, + stdout=json.dumps({"files": [{"path": "lib/foo.py"}]}), + ) + return MagicMock(returncode=1, stdout="") + + mock_run.side_effect = side_effect + + cmd_scan(db, "alice", []) + capsys.readouterr() + + alice_msgs = db.query("SELECT body FROM messages WHERE recipient='alice' AND body LIKE '%CI FAIL%'") + bob_msgs = db.query("SELECT body FROM messages WHERE recipient='bob' AND body LIKE '%CI FAIL%'") + assert len(alice_msgs) == 1 + assert len(bob_msgs) == 0 + assert "lib/foo.py" in alice_msgs[0][0] + + @patch("lib.board_own.subprocess.run") + def test_no_files_falls_back_to_all_not_broadcast(self, mock_run, db, capsys): + env = MagicMock() + env.project_root = Path("/tmp/fake") + db.env = env + cmd_own(db, "alice", ["claim", "lib/"]) + cmd_own(db, "bob", ["claim", "tests/"]) + capsys.readouterr() + + def side_effect(cmd, **_): + if "issue" in cmd: + return MagicMock(returncode=0, stdout="[]") + if "run" in cmd and "list" in cmd: + return MagicMock( + returncode=0, + stdout=json.dumps( + [{"status": "completed", "conclusion": "failure", "headBranch": "f2", "databaseId": 2}] + ), + ) + if "run" in cmd and "view" in cmd: + # No files → fallback path + return MagicMock(returncode=0, stdout=json.dumps({"files": []})) + return MagicMock(returncode=1, stdout="") + + mock_run.side_effect = side_effect + + cmd_scan(db, "alice", []) + capsys.readouterr() + + # ONE fallback message, not one per owner (the old broadcast was the bug) + all_msgs = db.query("SELECT body FROM messages WHERE body LIKE '%CI FAIL%'") + assert len(all_msgs) == 1 + assert "无法定位 owner" in all_msgs[0][0] + + @patch("lib.board_own.subprocess.run") + def test_gh_run_view_timeout_uses_fallback(self, mock_run, db, capsys): + env = MagicMock() + env.project_root = Path("/tmp/fake") + db.env = env + cmd_own(db, "alice", ["claim", "lib/"]) + capsys.readouterr() + + import subprocess as sp + + def side_effect(cmd, **_): + if "issue" in cmd: + return MagicMock(returncode=0, stdout="[]") + if "run" in cmd and "list" in cmd: + return MagicMock( + returncode=0, + stdout=json.dumps( + [{"status": "completed", "conclusion": "failure", "headBranch": "f3", "databaseId": 3}] + ), + ) + if "run" in cmd and "view" in cmd: + raise sp.TimeoutExpired(cmd, 15) + return MagicMock(returncode=1, stdout="") + + mock_run.side_effect = side_effect + + cmd_scan(db, "alice", []) + capsys.readouterr() + msgs = db.query("SELECT recipient, body FROM messages WHERE body LIKE '%CI FAIL%'") + assert len(msgs) == 1 + # fallback recipient is "all" (no lead in fixture) + assert msgs[0][0] == "all" + assert "无法定位 owner" in msgs[0][1] + + +# --------------------------------------------------------------------------- +# Audit subcommand +# --------------------------------------------------------------------------- + + +class TestOwnAudit: + def test_audit_shows_recent_entries(self, db, capsys): + from lib.board_own import _record_routing + + _record_routing(db, "issue", "#1", RouteDecision("alice", "assigned:alice", "high")) + _record_routing(db, "ci", "branch:99", RouteDecision("bob", "files:lib/x.py", "high")) + + cmd_own(db, "alice", ["audit"]) + out = capsys.readouterr().out + assert "alice" in out + assert "assigned:alice" in out + assert "bob" in out + assert "files:lib/x.py" in out + + def test_audit_empty(self, db, capsys): + cmd_own(db, "alice", ["audit"]) + out = capsys.readouterr().out + assert "为空" in out + + def test_audit_limit_flag(self, db, capsys): + from lib.board_own import _record_routing + + for i in range(5): + _record_routing(db, "issue", f"#{i}", RouteDecision("alice", "assigned:alice", "high")) + + cmd_own(db, "alice", ["audit", "--limit", "2"]) + out = capsys.readouterr().out + # Header says "最近 2 条", and 2 entry lines (each starts with " [") + assert "最近 2 条" in out + assert out.count("\n [") == 2 + + def test_audit_invalid_limit_exits(self, db): + with pytest.raises(SystemExit): + cmd_own(db, "alice", ["audit", "--limit", "abc"])