Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
32f58712e5 | ||
|
|
6865ae718a | ||
|
|
3e2f7a7ddb | ||
|
|
194f5e28bd | ||
|
|
fc7d238d06 | ||
|
|
66e21dcd5c | ||
|
|
e2157998d7 | ||
|
|
d9ea175a17 | ||
|
|
b67d9f233d | ||
|
|
608da13c0f | ||
|
|
c2ab3f39cf |
No files matched your search
@@ -1,4 +1,4 @@
|
||||
.PHONY: install project test update
|
||||
.PHONY: install project test update budget
|
||||
|
||||
# Install the skills globally into $HOME (.agents/skills + .claude/skills).
|
||||
install:
|
||||
@@ -19,3 +19,7 @@ test:
|
||||
# Update all skills to their latest versions and refresh skills-lock.json.
|
||||
update:
|
||||
@bash scripts/update.sh
|
||||
|
||||
# Count skill description words against the listing budget: make budget [INDEX=1]
|
||||
budget:
|
||||
@python3 scripts/budget.py --repo . --top 5 $(if $(INDEX),--index)
|
||||
@@ -9,7 +9,7 @@ Compatible with Claude Code, Codex, Gemini CLI, Cursor, Copilot, Windsurf, Cline
|
||||
|
||||
## Naming 🏷️
|
||||
|
||||
Every first-party skill is named `kvb-<category>-<name>` with one of these categories: `think`, `create`, `code`, `ci`, `git`, `check`, `org`, `skill`, `coach`. Only `kvb-cleanup`, `kvb-coach`, `kvb-help` and the collectors `kvb-dream`, `kvb-optimize` and `kvb-transcend` stand without a category. Names use `a-z`, `0-9` and single hyphens, at most 64 characters, and equal their directory name. `tests/test_naming.py` enforces all of it, including that the organigram below lists every skill under its category.
|
||||
Every first-party skill is named `kvb-<category>-<name>` with one of these categories: `think`, `create`, `code`, `ci`, `git`, `check`, `org`, `skill`, `coach`: `think` thinks without producing a file, `create` produces an artifact, `code` works on code, `ci` on pipelines, `git` on history and forges, `check` analyses and reports findings, `org` coordinates agents, `skill` works on skills themselves, `coach` serves the operator's wellbeing. Only `kvb-cleanup`, `kvb-coach`, `kvb-help` and the collectors `kvb-dream`, `kvb-optimize` and `kvb-transcend` stand without a category. Names use `a-z`, `0-9` and single hyphens, at most 64 characters, and equal their directory name. `tests/test_naming.py` enforces all of it, including that the organigram below lists every skill under its category.
|
||||
|
||||
## Organigram 🏛️
|
||||
|
||||
@@ -49,6 +49,8 @@ flowchart LR
|
||||
git --> git_commit_next[kvb-git-commit-next]
|
||||
git --> git_commit_no_verify[kvb-git-commit-no-verify]
|
||||
git --> git_merge_fix[kvb-git-merge-fix]
|
||||
git --> git_close_issues[kvb-git-close-issues]
|
||||
git --> git_update_branches[kvb-git-update-branches]
|
||||
git --> git_branch_gc[kvb-git-branch-gc]
|
||||
git --> git_changelog[kvb-git-changelog]
|
||||
check --> check_legal[kvb-check-legal]
|
||||
@@ -126,6 +128,8 @@ flowchart LR
|
||||
commit_next[kvb-git-commit-next]
|
||||
commit_no_verify[kvb-git-commit-no-verify]
|
||||
merge_fix[kvb-git-merge-fix]
|
||||
close_issues[kvb-git-close-issues]
|
||||
update_branches[kvb-git-update-branches]
|
||||
branch_gc[kvb-git-branch-gc]
|
||||
changelog[kvb-git-changelog]
|
||||
end
|
||||
@@ -182,6 +186,10 @@ flowchart LR
|
||||
commit_next --> commit_no_verify
|
||||
commit_next --> explain
|
||||
commit_next --> dialectic
|
||||
close_issues --> confidence
|
||||
update_branches --> active_listening
|
||||
update_branches -. conflicts .-> merge_fix
|
||||
update_branches -. pushes .-> operator
|
||||
commit_no_verify --> comments_clean
|
||||
commit_no_verify --> advocate
|
||||
advocate --> dialectic
|
||||
|
||||
Executable
+72
@@ -0,0 +1,72 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Count every first-party skill description against the listing budget.
|
||||
|
||||
Args:
|
||||
--repo: skills repository root.
|
||||
--index: count the staged versions in the git index instead of the files.
|
||||
--top: how many of the longest descriptions to list.
|
||||
|
||||
Exit codes:
|
||||
0 within budget, 1 over budget, 2 usage error.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
LISTING_WORDS = 1500
|
||||
DESCRIPTION_WORDS = 40
|
||||
|
||||
|
||||
def words(skill_md: str) -> int:
|
||||
frontmatter = skill_md.split("---", 2)[1]
|
||||
return len(frontmatter.split("description:", 1)[1].split()) - 1
|
||||
|
||||
|
||||
def git(repo: Path, *args: str) -> str:
|
||||
result = subprocess.run(
|
||||
["git", *args], cwd=repo, capture_output=True, text=True, check=False
|
||||
)
|
||||
if result.returncode != 0:
|
||||
print(f"budget: {result.stderr.strip()}", file=sys.stderr)
|
||||
sys.exit(2)
|
||||
return result.stdout
|
||||
|
||||
|
||||
def counts(repo: Path, index: bool) -> dict[str, int]:
|
||||
if index:
|
||||
paths = git(repo, "ls-files", "skills/*/SKILL.md").split()
|
||||
return {Path(p).parent.name: words(git(repo, "show", f":{p}")) for p in paths}
|
||||
return {
|
||||
p.parent.name: words(p.read_text(encoding="utf-8"))
|
||||
for p in sorted((repo / "skills").glob("*/SKILL.md"))
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
||||
parser.add_argument("--repo", required=True)
|
||||
parser.add_argument("--index", action="store_true")
|
||||
parser.add_argument("--top", type=int, required=True)
|
||||
args = parser.parse_args()
|
||||
|
||||
repo = Path(args.repo)
|
||||
if not (repo / "skills").is_dir():
|
||||
print(f"budget: {repo} has no skills/ directory", file=sys.stderr)
|
||||
return 2
|
||||
found = counts(repo, args.index)
|
||||
total = sum(found.values())
|
||||
over = sorted(name for name, n in found.items() if n > DESCRIPTION_WORDS)
|
||||
print(f"total {total}/{LISTING_WORDS} words in {len(found)} skills")
|
||||
for name, n in sorted(found.items(), key=lambda item: -item[1])[: args.top]:
|
||||
print(f"{n}\t{name}")
|
||||
for name in over:
|
||||
print(f"over {DESCRIPTION_WORDS}: {name}")
|
||||
return 1 if total > LISTING_WORDS or over else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
+1
-1
@@ -30,6 +30,6 @@ fi
|
||||
log "skills: updating from ${lockfile}..."
|
||||
jq -r '.skills[].source' "${lockfile}" | sort -u | while read -r source; do
|
||||
log "skills: updating ${source}..."
|
||||
npx --yes skills add --yes --agent claude-code "${source}" </dev/null
|
||||
npx --yes skills add "${source}" --yes --agent claude-code </dev/null
|
||||
done
|
||||
log "skills: update complete."
|
||||
@@ -2,9 +2,8 @@
|
||||
name: kvb-code-no-defaults
|
||||
description: >
|
||||
Forbid default values for parameters, settings and environment lookups;
|
||||
demand explicit values and fail loudly. Apply AUTOMATICALLY on every task
|
||||
that writes, edits, generates, refactors or reviews code, config, templates,
|
||||
schemas or infrastructure.
|
||||
demand explicit values. Apply AUTOMATICALLY whenever code, config,
|
||||
templates or infrastructure are written, edited or reviewed.
|
||||
---
|
||||
|
||||
A default value is a guess, written at the point where the caller's intent is
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
---
|
||||
name: kvb-code-refactor
|
||||
description: >
|
||||
Refactor every uncommitted file (staged, unstaged, untracked) up to the
|
||||
project's standards without changing behaviour. Trigger on /kvb-code-refactor or when
|
||||
the operator asks to clean up, tidy or polish the current changes.
|
||||
Refactor every uncommitted file up to the project's standards without
|
||||
changing behaviour. Trigger on /kvb-code-refactor or asked to clean up,
|
||||
tidy or polish the current changes.
|
||||
---
|
||||
|
||||
The unit of work is the uncommitted change set, not the repository. Behaviour
|
||||
|
||||
@@ -117,7 +117,9 @@ improvement. No repository file changes before an answer comes back.
|
||||
|
||||
With the dialectic done and memory written, run the `kvb-optimize` skill and
|
||||
hand it this run's findings as its evidence, so none of its skills
|
||||
re-collects them.
|
||||
re-collects them. Its pass includes the `kvb-transcend` skill, which carries
|
||||
the scripts, tests and skills of every session in scope into their
|
||||
repositories.
|
||||
|
||||
## Step 8: clean up
|
||||
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
---
|
||||
name: kvb-git-close-issues
|
||||
description: >
|
||||
Close the issues a pull request fixes, each verified and confirmed.
|
||||
Trigger on /kvb-git-close-issues [pr].
|
||||
---
|
||||
|
||||
Closing an issue is outward-facing: nothing is closed or edited before the
|
||||
operator confirms the list, and every closed issue names the pull request
|
||||
and commit that fixed it.
|
||||
|
||||
## Step 1: candidates
|
||||
|
||||
The pull request named after the command, else the one of the current
|
||||
branch (`gh pr view --json number`). `${CLAUDE_SKILL_DIR}/scripts/issues.py`
|
||||
lists every issue it references, from closing keywords and mentions in the
|
||||
body and the commits, a trailing number in the branch name, and the issues
|
||||
GitHub already links:
|
||||
|
||||
```bash
|
||||
gh pr view <pr> --json number,body,headRefName,commits,closingIssuesReferences | issues.py --repo <owner/name>
|
||||
```
|
||||
|
||||
## Step 2: verify
|
||||
|
||||
Per candidate, `gh issue view <n> --json state,title,body,comments` (with
|
||||
`--repo` for a foreign one); a closed issue is skipped. Judge it `fixed`,
|
||||
`partly` or `unrelated` against the pull request's diff, with evidence: the
|
||||
change that addresses each point of the issue and the test or CI run that
|
||||
proves it. Only `fixed` goes on; `partly` gets a progress comment instead.
|
||||
|
||||
## Step 3: confirm
|
||||
|
||||
`AskUserQuestion` with `multiSelect: true`, one option per `fixed` issue,
|
||||
the evidence in its description, nothing preselected.
|
||||
|
||||
## Step 4: close
|
||||
|
||||
Where the fix lives decides how:
|
||||
|
||||
- **Merged into the default branch**: `gh issue close <n> --reason completed
|
||||
--comment "Fixed by #<pr> (<sha>)."`.
|
||||
- **Open, or merged into another branch**: add one `Closes #<n>` line per
|
||||
issue to the pull request body (`gh pr edit <pr> --body-file <file>`,
|
||||
keeping the existing body), so the merge into the default branch closes
|
||||
them; never close them early.
|
||||
|
||||
## Output
|
||||
|
||||
| Issue | Verdict | Evidence | Action |
|
||||
+95
@@ -0,0 +1,95 @@
|
||||
#!/usr/bin/env python3
|
||||
"""List the issues a pull request references, from `gh pr view --json` on stdin.
|
||||
|
||||
Args:
|
||||
stdin: JSON with number, body, headRefName, commits and
|
||||
closingIssuesReferences, as `gh pr view <pr> --json
|
||||
number,body,headRefName,commits,closingIssuesReferences` prints it.
|
||||
--repo: owner/name of the pull request's repository; its own references
|
||||
are reported without a repository.
|
||||
--json: print the candidates as JSON instead of tab-separated lines.
|
||||
|
||||
Exit codes:
|
||||
0 done, 2 usage error.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
|
||||
KEYWORD = r"(?:close[sd]?|fix(?:e[sd])?|resolve[sd]?)"
|
||||
REF = r"(?:(?P<repo>[\w.-]+/[\w.-]+)#|https://github\.com/(?P<url>[\w.-]+/[\w.-]+)/issues/|(?<![\w/])#)(?P<num>[1-9]\d*)"
|
||||
PATTERN = re.compile(rf"(?P<kw>\b{KEYWORD}\b:?\s+)?{REF}", re.IGNORECASE)
|
||||
BRANCH = re.compile(r"[/-](\d+)$")
|
||||
|
||||
|
||||
def fail(message: str) -> None:
|
||||
print(f"issues: {message}", file=sys.stderr)
|
||||
sys.exit(2)
|
||||
|
||||
|
||||
def scan(text: str, source: str, found: dict) -> None:
|
||||
for match in PATTERN.finditer(text or ""):
|
||||
repo = match.group("repo") or match.group("url") or ""
|
||||
entry = found.setdefault(
|
||||
(repo, int(match.group("num"))), {"closing": False, "sources": []}
|
||||
)
|
||||
entry["closing"] = entry["closing"] or bool(match.group("kw"))
|
||||
if source not in entry["sources"]:
|
||||
entry["sources"].append(source)
|
||||
|
||||
|
||||
def candidates(pr: dict, own: str) -> list[dict]:
|
||||
found: dict = {}
|
||||
scan(pr.get("body", ""), "body", found)
|
||||
for commit in pr.get("commits", []):
|
||||
text = f"{commit.get('messageHeadline', '')}\n{commit.get('messageBody', '')}"
|
||||
scan(text, f"commit {commit.get('oid', '')[:10]}", found)
|
||||
branch = BRANCH.search(pr.get("headRefName", ""))
|
||||
if branch:
|
||||
found.setdefault(("", int(branch.group(1))), {"closing": False, "sources": []})[
|
||||
"sources"
|
||||
].append("branch")
|
||||
for linked in pr.get("closingIssuesReferences", []):
|
||||
url = re.search(r"github\.com/([\w.-]+/[\w.-]+)/issues/", linked.get("url", ""))
|
||||
key = (url.group(1) if url else "", int(linked["number"]))
|
||||
entry = found.setdefault(key, {"closing": False, "sources": []})
|
||||
entry["closing"] = True
|
||||
entry["sources"].append("linked")
|
||||
merged: dict = {}
|
||||
for (repo, num), entry in found.items():
|
||||
key = ("" if repo.lower() == own.lower() else repo, num)
|
||||
if key == ("", pr.get("number")):
|
||||
continue
|
||||
target = merged.setdefault(key, {"closing": False, "sources": []})
|
||||
target["closing"] = target["closing"] or entry["closing"]
|
||||
target["sources"] += [s for s in entry["sources"] if s not in target["sources"]]
|
||||
return [{"repo": repo, "issue": num, **entry} for (repo, num), entry in sorted(merged.items())]
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
||||
parser.add_argument("--repo", required=True)
|
||||
parser.add_argument("--json", action="store_true")
|
||||
args = parser.parse_args()
|
||||
try:
|
||||
pr = json.load(sys.stdin)
|
||||
except ValueError:
|
||||
fail("stdin is not the JSON of gh pr view --json")
|
||||
if not isinstance(pr, dict):
|
||||
fail("stdin is not a pull request object")
|
||||
rows = candidates(pr, args.repo)
|
||||
if args.json:
|
||||
print(json.dumps(rows, indent=1))
|
||||
return 0
|
||||
for row in rows:
|
||||
kind = "closing" if row["closing"] else "mention"
|
||||
print(f"{row['repo'] or '-'}\t{row['issue']}\t{kind}\t{', '.join(row['sources'])}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -29,7 +29,9 @@ description: >
|
||||
see tracked, modified files and read the worktree; only an untracked
|
||||
NEW file needs `git add` to become visible. Path-scope every `git add`
|
||||
to the bundle — `git add -A` sweeps up the operator's own edits and
|
||||
their local config.
|
||||
their local config. A file that carries two bundles is staged by line:
|
||||
`${CLAUDE_SKILL_DIR}/scripts/stage_hunks.py --repo <worktree> --file
|
||||
<path> --match <regex>`, first with `--dry-run`.
|
||||
5. For the newly staged bundle run the kvb-think-explain skill (what the changes do
|
||||
and how the pieces interact, file:line-anchored) followed by the
|
||||
kvb-check-confidence skill (fix confidence, break risk, evidence, residuals).
|
||||
|
||||
+95
@@ -0,0 +1,95 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Stage only the changed lines of one file that match a pattern.
|
||||
|
||||
Args:
|
||||
--repo: path of the git worktree.
|
||||
--file: the file, relative to the worktree root.
|
||||
--match: regular expression; a hunk of only added lines keeps the added
|
||||
lines it matches, any other hunk is staged whole when one of its
|
||||
changed lines matches.
|
||||
--dry-run: print the patch instead of applying it to the index.
|
||||
|
||||
Exit codes:
|
||||
0 staged or printed, 1 nothing matched, 2 usage error.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def fail(message: str) -> None:
|
||||
print(f"stage_hunks: {message}", file=sys.stderr)
|
||||
sys.exit(2)
|
||||
|
||||
|
||||
def git(repo: Path, *args: str, stdin: str | None = None) -> str:
|
||||
result = subprocess.run(
|
||||
["git", *args], cwd=repo, input=stdin, capture_output=True, text=True, check=False
|
||||
)
|
||||
if result.returncode != 0:
|
||||
fail(result.stderr.strip() or f"git {' '.join(args)} failed")
|
||||
return result.stdout
|
||||
|
||||
|
||||
def split(diff: str) -> tuple[list[str], list[list[str]]]:
|
||||
header: list[str] = []
|
||||
hunks: list[list[str]] = []
|
||||
for line in diff.splitlines():
|
||||
if line.startswith("@@"):
|
||||
hunks.append([line])
|
||||
elif hunks:
|
||||
hunks[-1].append(line)
|
||||
else:
|
||||
header.append(line)
|
||||
return header, hunks
|
||||
|
||||
|
||||
def select(hunks: list[list[str]], pattern: re.Pattern) -> list[list[str]]:
|
||||
kept = []
|
||||
for hunk in hunks:
|
||||
changes = [line for line in hunk[1:] if line[:1] in "+-"]
|
||||
if all(line.startswith("+") for line in changes):
|
||||
lines = [line for line in changes if pattern.search(line[1:])]
|
||||
if lines:
|
||||
kept.append([hunk[0], *lines])
|
||||
elif any(pattern.search(line[1:]) for line in changes):
|
||||
kept.append(hunk)
|
||||
return kept
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
||||
parser.add_argument("--repo", required=True)
|
||||
parser.add_argument("--file", required=True)
|
||||
parser.add_argument("--match", required=True)
|
||||
parser.add_argument("--dry-run", action="store_true")
|
||||
args = parser.parse_args()
|
||||
|
||||
repo = Path(args.repo)
|
||||
if not repo.is_dir():
|
||||
fail(f"{repo} is not a directory")
|
||||
try:
|
||||
pattern = re.compile(args.match)
|
||||
except re.error as error:
|
||||
fail(f"--match: {error}")
|
||||
header, hunks = split(git(repo, "diff", "--no-color", "-U0", "--", args.file))
|
||||
kept = select(hunks, pattern)
|
||||
if not kept:
|
||||
print(f"stage_hunks: no change in {args.file} matches {args.match!r}", file=sys.stderr)
|
||||
return 1
|
||||
patch = "\n".join([*header, *(line for hunk in kept for line in hunk)]) + "\n"
|
||||
if args.dry_run:
|
||||
print(patch, end="")
|
||||
return 0
|
||||
git(repo, "apply", "--cached", "--unidiff-zero", "--recount", "-", stdin=patch)
|
||||
print(f"staged {sum(len(h) - 1 for h in kept)} lines of {args.file}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,52 @@
|
||||
---
|
||||
name: kvb-git-update-branches
|
||||
description: >
|
||||
Merge main into open PRs or branches: the operator's own directly, others
|
||||
after consent. Trigger on /kvb-git-update-branches.
|
||||
---
|
||||
|
||||
Invoking this skill authorizes the merge commits it makes; pushing stays the
|
||||
operator's. Branches are merged, never rebased.
|
||||
|
||||
## Step 1: scope and candidates
|
||||
|
||||
Pull requests, branches or both, as the operator said; ask when it is open.
|
||||
`git fetch --all --prune`, then list the open branches with their lag and
|
||||
owner:
|
||||
|
||||
```bash
|
||||
gh pr list --state open --json number,headRefName,author,isCrossRepository > <tmp>/prs.json
|
||||
branches.py --repo <repo> --remote origin --base main --email "<git user.email>" --login "<gh login>" --prs <tmp>/prs.json
|
||||
```
|
||||
|
||||
`${CLAUDE_SKILL_DIR}/scripts/branches.py` drops branches already merged into
|
||||
the base and marks `mine`: a pull request by the operator's login, or a
|
||||
branch whose own commits all carry the operator's email. A row with
|
||||
`behind 0` is current and skipped.
|
||||
|
||||
## Step 2: consent
|
||||
|
||||
The operator's own rows go on. For every other row run the `kvb-think-listen`
|
||||
skill: one `AskUserQuestion` with `multiSelect: true` listing branch, pull
|
||||
request, owner and lag, nothing preselected. Only what the operator picks is
|
||||
updated.
|
||||
|
||||
## Step 3: update
|
||||
|
||||
- **Fork pull request**: no local merge; its `gh pr update-branch <n>` goes
|
||||
to the operator.
|
||||
- **Any other branch**: in the worktree that already has it checked out,
|
||||
else a temporary one under `/tmp` (the repository's own worktree targets
|
||||
where it has them), `git merge --no-verify --no-edit origin/main`. On a
|
||||
conflict, `git merge --abort` and report the files; the `kvb-git-merge-fix`
|
||||
skill or the operator resolves it later.
|
||||
- Remove the temporary worktrees afterwards.
|
||||
|
||||
## Step 4: hand over the pushes
|
||||
|
||||
Every updated branch's push and every `gh pr update-branch` goes to the
|
||||
`kvb-org-operator` skill as an operator command.
|
||||
|
||||
## Output
|
||||
|
||||
| Branch | PR | Owner | Behind | Result | Push command |
|
||||
+121
@@ -0,0 +1,121 @@
|
||||
#!/usr/bin/env python3
|
||||
"""List the open branches of a remote with their lag behind the base and their owner.
|
||||
|
||||
Args:
|
||||
--repo: path of the local clone.
|
||||
--remote: remote whose branches are listed, e.g. origin.
|
||||
--base: base branch on that remote, e.g. main.
|
||||
--email: the operator's commit email; a branch whose own commits all carry
|
||||
it belongs to the operator.
|
||||
--login: the operator's forge login, matched against pull request authors.
|
||||
--prs: file with `gh pr list --state open --json
|
||||
number,headRefName,author,isCrossRepository`; omit it for branches only.
|
||||
--json: print the rows as JSON instead of tab-separated lines.
|
||||
|
||||
Exit codes:
|
||||
0 done, 2 usage error.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def fail(message: str) -> None:
|
||||
print(f"branches: {message}", file=sys.stderr)
|
||||
sys.exit(2)
|
||||
|
||||
|
||||
def git(repo: Path, *args: str) -> str:
|
||||
result = subprocess.run(
|
||||
["git", *args], cwd=repo, capture_output=True, text=True, check=False
|
||||
)
|
||||
if result.returncode != 0:
|
||||
fail(result.stderr.strip() or f"git {' '.join(args)} failed")
|
||||
return result.stdout
|
||||
|
||||
|
||||
def load_prs(path: str | None) -> list[dict]:
|
||||
if path is None:
|
||||
return []
|
||||
try:
|
||||
prs = json.loads(Path(path).read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError):
|
||||
fail(f"{path} is not the JSON of gh pr list")
|
||||
if not isinstance(prs, list):
|
||||
fail(f"{path} is not a list of pull requests")
|
||||
return prs
|
||||
|
||||
|
||||
def rows(repo: Path, remote: str, base: str, email: str, login: str, prs: list[dict]) -> list[dict]:
|
||||
target = f"{remote}/{base}"
|
||||
git(repo, "rev-parse", "--verify", target)
|
||||
by_branch = {pr["headRefName"]: pr for pr in prs if not pr.get("isCrossRepository")}
|
||||
found = []
|
||||
refs = git(repo, "for-each-ref", "--format=%(refname:short)", f"refs/remotes/{remote}/")
|
||||
for ref in refs.split():
|
||||
branch = ref.removeprefix(f"{remote}/")
|
||||
if branch in (base, "HEAD") or ref == remote:
|
||||
continue
|
||||
ahead = int(git(repo, "rev-list", "--count", f"{target}..{ref}"))
|
||||
if ahead == 0:
|
||||
continue
|
||||
behind = int(git(repo, "rev-list", "--count", f"{ref}..{target}"))
|
||||
authors = sorted(set(git(repo, "log", "--format=%ae", f"{target}..{ref}").split()))
|
||||
pr = by_branch.get(branch)
|
||||
mine = (pr["author"]["login"] == login) if pr else authors == [email]
|
||||
found.append(
|
||||
{
|
||||
"branch": branch,
|
||||
"pr": pr["number"] if pr else None,
|
||||
"ahead": ahead,
|
||||
"behind": behind,
|
||||
"authors": authors,
|
||||
"mine": mine,
|
||||
}
|
||||
)
|
||||
for pr in prs:
|
||||
if pr.get("isCrossRepository"):
|
||||
found.append(
|
||||
{
|
||||
"branch": pr["headRefName"],
|
||||
"pr": pr["number"],
|
||||
"ahead": None,
|
||||
"behind": None,
|
||||
"authors": [pr["author"]["login"]],
|
||||
"mine": pr["author"]["login"] == login,
|
||||
"fork": True,
|
||||
}
|
||||
)
|
||||
return found
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
||||
for name in ("--repo", "--remote", "--base", "--email", "--login"):
|
||||
parser.add_argument(name, required=True)
|
||||
parser.add_argument("--prs")
|
||||
parser.add_argument("--json", action="store_true")
|
||||
args = parser.parse_args()
|
||||
|
||||
repo = Path(args.repo)
|
||||
if not repo.is_dir():
|
||||
fail(f"{repo} is not a directory")
|
||||
found = rows(repo, args.remote, args.base, args.email, args.login, load_prs(args.prs))
|
||||
if args.json:
|
||||
print(json.dumps(found, indent=1))
|
||||
return 0
|
||||
for row in found:
|
||||
print(
|
||||
f"{row['branch']}\t#{row['pr'] or '-'}\tahead {row['ahead']}\tbehind {row['behind']}"
|
||||
f"\t{'mine' if row['mine'] else ', '.join(row['authors'])}"
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -1,10 +1,9 @@
|
||||
---
|
||||
name: kvb-org-organize
|
||||
description: >
|
||||
Keep parallel agents from duplicating work: the first claim owns it; a
|
||||
resource chair allocates. Run AUTOMATICALLY
|
||||
before work another agent might also do (failure, cluster, feature, gate,
|
||||
deploy, cleanup), and on /kvb-org-organize.
|
||||
Keep parallel agents from duplicating work: the first claim owns it, a
|
||||
chair allocates. Run AUTOMATICALLY before work another agent might also
|
||||
do, and on /kvb-org-organize.
|
||||
---
|
||||
|
||||
Two agents doing the same work waste the second run and then collide in the
|
||||
|
||||
@@ -4,7 +4,8 @@
|
||||
Environment:
|
||||
AGENTS_COUNCIL_ROOT: council directory shared by every agent on the host;
|
||||
/tmp/agents-council unless set, because all agents must agree on it.
|
||||
CLAUDE_CODE_SESSION_ID: the caller's session; other runtimes pass --session.
|
||||
CLAUDE_CODE_SESSION_ID: the caller's session, taken over --session; other
|
||||
runtimes pass --session.
|
||||
|
||||
Exit codes:
|
||||
0 done, 2 usage error, 3 office held by a live session,
|
||||
@@ -46,11 +47,11 @@ def live_sessions() -> dict[str, str | None]:
|
||||
|
||||
|
||||
def caller(session: str | None) -> tuple[str, str]:
|
||||
if session:
|
||||
return session, "other"
|
||||
env = os.environ.get("CLAUDE_CODE_SESSION_ID")
|
||||
if env:
|
||||
return env, "claude-code"
|
||||
if session:
|
||||
return session, "other"
|
||||
print("council: no session id; pass --session", file=sys.stderr)
|
||||
sys.exit(2)
|
||||
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
---
|
||||
name: kvb-org-perform
|
||||
description: >
|
||||
Pick `model` and `effort` per subagent from a fixed table: cheaper only at
|
||||
90 % confidence that the result equals the default's; log spawns and
|
||||
quality. Apply AUTOMATICALLY before every Agent tool call.
|
||||
Pick `model` and `effort` per subagent from a fixed table, cheaper only at
|
||||
90 % confidence of an equal result. Apply AUTOMATICALLY before every Agent
|
||||
tool call.
|
||||
---
|
||||
|
||||
A subagent runs on the main model and effort unless the spawn names others.
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
---
|
||||
name: kvb-think-dialectic
|
||||
description: >
|
||||
Reach a defensible conclusion by thesis, independent skeptic antithesis and
|
||||
synthesis, iterated to ~99% confidence. Trigger for any consequential root
|
||||
cause, design or decision that is uncertain and expensive to get wrong.
|
||||
Thesis, independent skeptic antithesis and synthesis, iterated to ~99%
|
||||
confidence. Trigger for any root cause, design or decision that is
|
||||
uncertain and expensive to get wrong.
|
||||
---
|
||||
|
||||
Reach the truth by dialectic, not by first guess. Do not stop at a plausible
|
||||
|
||||
@@ -19,7 +19,9 @@ transcend.py mark --state <same> <path> carried|dropped|later
|
||||
scratchpad, every peer session's scratchpad, and the agent homes
|
||||
(`~/.claude`, `~/.agents`, `~/.codex`) for skills, commands, agents, hooks
|
||||
and rules the skills repository does not ship. It keeps scripts, tests and
|
||||
skills, and skips files named like secrets and files over 1 MB. Carried or
|
||||
skills at most one directory below a scratchpad, and skips files named like
|
||||
secrets, files over 1 MB, and repository copies (a directory holding `.git`,
|
||||
`roles`, `plugins`, `pyproject.toml`, `package.json` or a `Makefile`). Carried or
|
||||
dropped candidates return only when their content changes; `later` returns
|
||||
every time.
|
||||
|
||||
|
||||
@@ -22,6 +22,8 @@ SECRET_NAME = re.compile(
|
||||
r"(secret|token|credential|passw|auth|\.env$|id_[a-z0-9]+$|\.pem$|\.key$)", re.IGNORECASE
|
||||
)
|
||||
MAX_BYTES = 1_000_000
|
||||
MAX_DEPTH = 1
|
||||
REPO_MARKERS = {".git", "roles", "plugins", "pyproject.toml", "package.json", "Makefile"}
|
||||
HOME_SOURCES = {
|
||||
".claude/skills": "skill",
|
||||
".agents/skills": "skill",
|
||||
@@ -74,11 +76,21 @@ def session_names(home: Path) -> dict[str, str]:
|
||||
|
||||
def scratch_candidates(root: Path, source: str, owner: str) -> list[dict]:
|
||||
found = []
|
||||
for path in sorted(p for p in root.rglob("*") if p.is_file()):
|
||||
kind = file_kind(path)
|
||||
if kind:
|
||||
found.append({"source": source, "owner": owner, "kind": kind, "path": str(path)})
|
||||
return found
|
||||
for current, dirs, files in os.walk(root):
|
||||
here = Path(current)
|
||||
depth = len(here.relative_to(root).parts)
|
||||
if depth and REPO_MARKERS & {*dirs, *files}:
|
||||
dirs.clear()
|
||||
continue
|
||||
if depth >= MAX_DEPTH:
|
||||
dirs.clear()
|
||||
for name in files:
|
||||
kind = file_kind(here / name)
|
||||
if kind:
|
||||
found.append(
|
||||
{"source": source, "owner": owner, "kind": kind, "path": str(here / name)}
|
||||
)
|
||||
return sorted(found, key=lambda item: item["path"])
|
||||
|
||||
|
||||
def shipped(skills_repo: Path) -> set[str]:
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
"""Exercise issues.py: closing keywords, mentions, commits, branch number and linked issues."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPT = REPO_ROOT / "skills" / "kvb-git-close-issues" / "scripts" / "issues.py"
|
||||
PR = {
|
||||
"number": 40,
|
||||
"body": (
|
||||
"Fixes #12 and touches #13.\nCloses: other/repo#7\nSee https://github.com/a/b/issues/9\n"
|
||||
"Rows WordPress#0 and docs#1 are not issues; this is #40."
|
||||
),
|
||||
"headRefName": "fix/sys-ctl-hlth-runtime/vulnerability-engines/1324",
|
||||
"commits": [
|
||||
{"oid": "abcdef1234567", "messageHeadline": "fix: guard", "messageBody": "Resolved #13"},
|
||||
{"oid": "1111111111111", "messageHeadline": "chore: tidy", "messageBody": ""},
|
||||
],
|
||||
"closingIssuesReferences": [
|
||||
{"number": 21, "url": "https://github.com/owner/repo/issues/21"}
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _run(payload: str) -> subprocess.CompletedProcess:
|
||||
return subprocess.run(
|
||||
[sys.executable, str(SCRIPT), "--repo", "Owner/repo", "--json"],
|
||||
input=payload,
|
||||
check=False,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
|
||||
|
||||
class TestIssues(unittest.TestCase):
|
||||
def test_every_reference_is_found_with_its_kind_and_source(self):
|
||||
result = _run(json.dumps(PR))
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
rows = {(r["repo"], r["issue"]): r for r in json.loads(result.stdout)}
|
||||
self.assertEqual(
|
||||
{key: (row["closing"], row["sources"]) for key, row in rows.items()},
|
||||
{
|
||||
("", 12): (True, ["body"]),
|
||||
("", 13): (True, ["body", "commit abcdef1234"]),
|
||||
("other/repo", 7): (True, ["body"]),
|
||||
("a/b", 9): (False, ["body"]),
|
||||
("", 1324): (False, ["branch"]),
|
||||
("", 21): (True, ["linked"]),
|
||||
},
|
||||
)
|
||||
|
||||
def test_invalid_input_exits_2(self):
|
||||
self.assertEqual(_run("not json").returncode, 2)
|
||||
self.assertEqual(_run("[]").returncode, 2)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -62,6 +62,11 @@ class TestCouncil(unittest.TestCase):
|
||||
self._run("s1", "claim", "work/x", "--work", "w")
|
||||
self.assertEqual(self._run("s2", "takeover", "work/x", "--work", "w").returncode, 3)
|
||||
|
||||
def test_session_flag_cannot_hide_the_claude_code_session(self):
|
||||
self._run("s1", "claim", "work/x", "--work", "w", "--session", "One")
|
||||
self.assertIn('"session": "s1"', (self.root / "work/x/claim.json").read_text())
|
||||
self.assertEqual(self._run("s1", "release", "work/x", "--session", "One").returncode, 0)
|
||||
|
||||
def test_only_the_holder_releases(self):
|
||||
self._run("s1", "claim", "work/x", "--work", "w")
|
||||
self.assertEqual(self._run("s2", "release", "work/x").returncode, 3)
|
||||
|
||||
+22
-10
@@ -2,18 +2,23 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
LOCK_FILE = REPO_ROOT / "skills-lock.json"
|
||||
BUDGET = REPO_ROOT / "scripts" / "budget.py"
|
||||
SPEC = importlib.util.spec_from_file_location("budget", BUDGET)
|
||||
budget = importlib.util.module_from_spec(SPEC)
|
||||
SPEC.loader.exec_module(budget)
|
||||
|
||||
HASH_RE = re.compile(r"^[0-9a-f]{64}$")
|
||||
REQUIRED_FIELDS = ("source", "sourceType", "skillPath", "computedHash")
|
||||
DESCRIPTION_WORDS = 40
|
||||
LISTING_WORDS = 1300
|
||||
|
||||
|
||||
class TestSkillsLock(unittest.TestCase):
|
||||
@@ -64,14 +69,21 @@ class TestSkillsLock(unittest.TestCase):
|
||||
)
|
||||
|
||||
def test_descriptions_fit_the_listing_budget(self):
|
||||
total = 0
|
||||
for skill_md in sorted((REPO_ROOT / "skills").glob("*/SKILL.md")):
|
||||
frontmatter = skill_md.read_text(encoding="utf-8").split("---", 2)[1]
|
||||
words = len(frontmatter.split("description:", 1)[1].split()) - 1
|
||||
total += words
|
||||
with self.subTest(skill=skill_md.parent.name):
|
||||
self.assertLessEqual(words, DESCRIPTION_WORDS)
|
||||
self.assertLessEqual(total, LISTING_WORDS)
|
||||
found = budget.counts(REPO_ROOT, index=False)
|
||||
for name, words in found.items():
|
||||
with self.subTest(skill=name):
|
||||
self.assertLessEqual(words, budget.DESCRIPTION_WORDS)
|
||||
self.assertLessEqual(sum(found.values()), budget.LISTING_WORDS)
|
||||
|
||||
def test_budget_script_reports_the_total_and_the_longest(self):
|
||||
result = subprocess.run(
|
||||
[sys.executable, str(BUDGET), "--repo", str(REPO_ROOT), "--top", "3"],
|
||||
capture_output=True, text=True, check=False,
|
||||
)
|
||||
self.assertEqual(result.returncode, 0, result.stdout)
|
||||
self.assertRegex(result.stdout, r"^total \d+/\d+ words in \d+ skills\n")
|
||||
self.assertEqual(len(result.stdout.splitlines()), 4)
|
||||
self.assertEqual(budget.words("---\nname: x\ndescription: >\n one two three\n---\n"), 3)
|
||||
|
||||
def test_first_party_skills_do_not_shadow_locked_skills(self):
|
||||
locked = set(self.lock["skills"].keys())
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
"""Exercise stage_hunks.py: matching added lines, whole changed hunks, the rest stays unstaged."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPT = REPO_ROOT / "skills" / "kvb-git-commit-next" / "scripts" / "stage_hunks.py"
|
||||
|
||||
|
||||
@unittest.skipUnless(shutil.which("git"), "git not installed")
|
||||
class TestStageHunks(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.repo = Path(self.tmp.name)
|
||||
self.env = {
|
||||
"PATH": "/usr/bin:/bin",
|
||||
"HOME": str(self.repo),
|
||||
"GIT_AUTHOR_NAME": "t",
|
||||
"GIT_AUTHOR_EMAIL": "t@t",
|
||||
"GIT_COMMITTER_NAME": "t",
|
||||
"GIT_COMMITTER_EMAIL": "t@t",
|
||||
}
|
||||
self.file = self.repo / "notes.txt"
|
||||
self.file.write_text("one\ntwo\nthree\nfour\nfive\n", encoding="utf-8")
|
||||
self._git("init", "-q")
|
||||
self._git("add", "notes.txt")
|
||||
self._git("commit", "-q", "-m", "base")
|
||||
self.file.write_text(
|
||||
"one\nalpha one\nbeta one\ntwo\nthree\nfour alpha\nfive\nbeta two\n", encoding="utf-8"
|
||||
)
|
||||
|
||||
def _git(self, *args: str) -> str:
|
||||
return subprocess.run(
|
||||
["git", *args], check=True, capture_output=True, text=True, env=self.env, cwd=self.repo
|
||||
).stdout
|
||||
|
||||
def _run(self, pattern: str) -> subprocess.CompletedProcess:
|
||||
return subprocess.run(
|
||||
[sys.executable, str(SCRIPT), "--repo", str(self.repo), "--file", "notes.txt", "--match", pattern],
|
||||
check=False, capture_output=True, text=True, env=self.env,
|
||||
)
|
||||
|
||||
def test_only_matching_changes_are_staged(self):
|
||||
result = self._run("alpha")
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
staged = [line for line in self._git("diff", "--cached", "-U0").splitlines() if line[:1] in "+-" and line[:3] not in ("+++", "---")]
|
||||
self.assertEqual(staged, ["+alpha one", "-four", "+four alpha"])
|
||||
unstaged = [line for line in self._git("diff", "-U0").splitlines() if line.startswith("+") and not line.startswith("+++")]
|
||||
self.assertEqual(unstaged, ["+beta one", "+beta two"])
|
||||
|
||||
def test_no_match_exits_1_and_bad_pattern_exits_2(self):
|
||||
self.assertEqual(self._run("gamma").returncode, 1)
|
||||
self.assertEqual(self._run("(").returncode, 2)
|
||||
self.assertEqual(self._git("diff", "--cached"), "")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -40,6 +40,10 @@ class TestTranscend(unittest.TestCase):
|
||||
self._write(self.own / "auth_token.py", "secret")
|
||||
self._write(self.own / "big.py", "x" * 1_000_001)
|
||||
self._write(self.peer / "helper.sh", "echo hi")
|
||||
self._write(self.peer / "probe" / "nested.py", "print(2)")
|
||||
self._write(self.peer / "probe" / "deep" / "too_deep.py", "print(3)")
|
||||
self._write(self.peer / "copy" / "pyproject.toml", "[project]")
|
||||
self._write(self.peer / "copy" / "copied.py", "print(4)")
|
||||
for name in ("kvb-x", "caveman", "custom"):
|
||||
self._write(self.home / ".claude" / "skills" / name / "SKILL.md", name)
|
||||
self._write(self.home / ".codex" / "rules" / "default.rules", "allow")
|
||||
@@ -78,6 +82,7 @@ class TestTranscend(unittest.TestCase):
|
||||
"probe.py": ("own", "self", "script"),
|
||||
"test_probe.py": ("own", "self", "test"),
|
||||
"helper.sh": ("peer", "Peer", "script"),
|
||||
"nested.py": ("peer", "Peer", "script"),
|
||||
"custom": ("home", "operator", "skill"),
|
||||
"default.rules": ("home", "operator", "rule"),
|
||||
},
|
||||
|
||||
@@ -36,10 +36,10 @@ class TestUpdate(unittest.TestCase):
|
||||
env={"PATH": f"{self.bin}:/usr/bin:/bin"},
|
||||
stdin=subprocess.DEVNULL,
|
||||
)
|
||||
calls = self.log.read_text(encoding="utf-8").splitlines()
|
||||
self.assertEqual([c.split()[-1] for c in calls], sources)
|
||||
calls = [c.split() for c in self.log.read_text(encoding="utf-8").splitlines()]
|
||||
self.assertEqual([c[c.index("add") + 1] for c in calls], sources)
|
||||
for call in calls:
|
||||
self.assertIn("--agent claude-code", call)
|
||||
self.assertEqual(call[call.index("--agent") + 1 :], ["claude-code"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -0,0 +1,92 @@
|
||||
"""Exercise branches.py: lag behind the base, ownership by commits and by PR author, forks."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPT = REPO_ROOT / "skills" / "kvb-git-update-branches" / "scripts" / "branches.py"
|
||||
ME = "me@example.org"
|
||||
|
||||
|
||||
@unittest.skipUnless(shutil.which("git"), "git not installed")
|
||||
class TestBranches(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
base = Path(self.tmp.name)
|
||||
self.env = {"PATH": "/usr/bin:/bin", "HOME": str(base)}
|
||||
self.origin = base / "origin.git"
|
||||
self.clone = base / "clone"
|
||||
self._git(base, "init", "-q", "--bare", "-b", "main", str(self.origin))
|
||||
self._git(base, "clone", "-q", str(self.origin), str(self.clone))
|
||||
self._commit(ME, "base")
|
||||
self._git(self.clone, "branch", "-M", "main")
|
||||
self._git(self.clone, "push", "-q", "origin", "main")
|
||||
for branch, email in (("feat/mine", ME), ("feat/theirs", "other@example.org"), ("feat/pr-mine", "other@example.org")):
|
||||
self._git(self.clone, "switch", "-q", "-c", branch, "main")
|
||||
self._commit(email, branch)
|
||||
self._git(self.clone, "push", "-q", "origin", branch)
|
||||
self._git(self.clone, "switch", "-q", "-c", "feat/merged", "main")
|
||||
self._git(self.clone, "push", "-q", "origin", "feat/merged")
|
||||
self._git(self.clone, "switch", "-q", "main")
|
||||
self._commit(ME, "main moves on")
|
||||
self._commit(ME, "and again")
|
||||
self._git(self.clone, "push", "-q", "origin", "main")
|
||||
self._git(self.clone, "fetch", "-q", "--prune")
|
||||
self.prs = base / "prs.json"
|
||||
self.prs.write_text(
|
||||
json.dumps(
|
||||
[
|
||||
{"number": 5, "headRefName": "feat/pr-mine", "author": {"login": "me"}, "isCrossRepository": False},
|
||||
{"number": 6, "headRefName": "fix", "author": {"login": "stranger"}, "isCrossRepository": True},
|
||||
]
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
def _git(self, cwd: Path, *args: str) -> None:
|
||||
subprocess.run(["git", *args], check=True, capture_output=True, env=self.env, cwd=cwd)
|
||||
|
||||
def _commit(self, email: str, message: str) -> None:
|
||||
env = {**self.env, "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": email, "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": email}
|
||||
subprocess.run(
|
||||
["git", "commit", "-q", "--allow-empty", "-m", message],
|
||||
check=True, capture_output=True, env=env, cwd=self.clone,
|
||||
)
|
||||
|
||||
def _run(self, *extra: str) -> subprocess.CompletedProcess:
|
||||
return subprocess.run(
|
||||
[sys.executable, str(SCRIPT), "--repo", str(self.clone), "--remote", "origin", "--base", "main",
|
||||
"--email", ME, "--login", "me", *extra],
|
||||
check=False, capture_output=True, text=True, env=self.env,
|
||||
)
|
||||
|
||||
def test_open_branches_carry_lag_and_owner(self):
|
||||
result = self._run("--prs", str(self.prs), "--json")
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
rows = {row["branch"]: row for row in json.loads(result.stdout)}
|
||||
self.assertEqual(set(rows), {"feat/mine", "feat/theirs", "feat/pr-mine", "fix"})
|
||||
self.assertEqual((rows["feat/mine"]["behind"], rows["feat/mine"]["mine"]), (2, True))
|
||||
self.assertEqual((rows["feat/theirs"]["mine"], rows["feat/theirs"]["pr"]), (False, None))
|
||||
self.assertEqual((rows["feat/pr-mine"]["pr"], rows["feat/pr-mine"]["mine"]), (5, True))
|
||||
self.assertEqual((rows["fix"]["mine"], rows["fix"].get("fork")), (False, True))
|
||||
|
||||
def test_invalid_input_exits_2(self):
|
||||
self.assertEqual(self._run("--prs", str(self.tmp.name) + "/missing.json").returncode, 2)
|
||||
broken = subprocess.run(
|
||||
[sys.executable, str(SCRIPT), "--repo", str(self.clone), "--remote", "origin", "--base", "nope",
|
||||
"--email", ME, "--login", "me"],
|
||||
check=False, capture_output=True, text=True, env=self.env,
|
||||
)
|
||||
self.assertEqual(broken.returncode, 2)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user