Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

feat: require workflow labels for issues (Closes #513)

Browse Source
This commit is contained in:
sysadmin
2026-07-08 04:20:24 -04:00
parent ce61e424f6
commit b7755f26e3
12 changed files with 838 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
create_issue.py
+53 -1
View File
@@ -29,6 +29,7 @@ from gitea_auth import (
get_credentials, resolve_remote, add_remote_args,
api_request, repo_api_url,
)
import issue_workflow_labels
def main(argv=None):
@@ -38,6 +39,16 @@ def main(argv=None):
parser.add_argument("--body", default="", help="Issue body text.")
parser.add_argument("--body-file",
help="Read issue body from this file ('-' for stdin).")
parser.add_argument("--label", action="append", default=[],
help="Existing label name to apply; may be repeated.")
parser.add_argument("--type-label",
help="Issue type, e.g. feature or type:feature.")
parser.add_argument("--status-label",
help="Initial status, e.g. ready or status:ready.")
parser.add_argument("--discussion", action="store_true",
help="Apply type:discussion.")
parser.add_argument("--require-workflow-labels", action="store_true",
help="Fail unless one type:* and one status:* label are supplied.")
args = parser.parse_args(argv)
host, org, repo = resolve_remote(args)
@@ -50,6 +61,24 @@ def main(argv=None):
with open(args.body_file, "r", encoding="utf-8") as fh:
body = fh.read()
requested_labels = issue_workflow_labels.labels_for_new_issue(
issue_type=args.type_label,
initial_status=args.status_label,
extra_labels=args.label,
discussion=args.discussion,
)
label_assessment = issue_workflow_labels.assess_issue_labels(
requested_labels,
discussion=args.discussion,
)
if args.require_workflow_labels and not label_assessment["valid"]:
print(
"Workflow label validation failed: "
+ "; ".join(label_assessment["errors"]),
file=sys.stderr,
)
return 1
user, password = get_credentials(host)
if not user or not password:
print(f"Could not get credentials for {host} "
@@ -59,11 +88,34 @@ def main(argv=None):
import base64
auth = f"Basic {base64.b64encode(f'{user}:{password}'.encode()).decode()}"
url = f"{repo_api_url(host, org, repo)}/issues"
base = repo_api_url(host, org, repo)
url = f"{base}/issues"
try:
label_ids = []
if requested_labels:
existing = api_request("GET", f"{base}/labels", auth)
by_name = {lb["name"]: lb["id"] for lb in existing}
missing = [name for name in requested_labels if name not in by_name]
if missing:
print(f"Missing labels: {missing}", file=sys.stderr)
return 1
label_ids = [by_name[name] for name in requested_labels]
data = api_request("POST", url, auth, {"title": args.title, "body": body})
if label_ids:
api_request(
"PUT",
f"{base}/issues/{data.get('number')}/labels",
auth,
{"labels": label_ids},
)
print(f"Issue #{data.get('number')}: {data.get('html_url')}")
if not label_assessment["valid"]:
print(
"Workflow label recommendation: "
+ "; ".join(label_assessment["errors"]),
file=sys.stderr,
)
return 0
except RuntimeError as e:
print(f"Error: {e}", file=sys.stderr)
docs/label-taxonomy.md
+88 -24
View File
@@ -1,34 +1,98 @@
# Label Taxonomy
This document catalogs the issue labels used for MCP workflows, including Jenkins and GlitchTip (observability).
This document defines the canonical issue labels used by MCP workflows.
> **Approval Required:** Do not create or apply new labels in `manage_labels.py` without explicit owner approval of this document.
Every issue should carry:
## Existing Labels
- one `type:*` label
- one `status:*` label
* **`jenkins`**
* Description: Jenkins integration
* Color: `d93f0b`
* Use: Used to mark issues, PRs, or tasks that involve the `jenkins-mcp` boundaries, CI/CD designs, or build failures.
Discussion-only issues must carry `type:discussion`.
* **`glitchtip`**
* Description: GlitchTip integration
* Color: `b60205`
* Use: Used to mark issues related to the `glitchtip-mcp` boundary and observability integration.
## Issue Type Labels
## Proposed / Missing Labels
| Label | Use |
| --- | --- |
| `type:bug` | Bug or defect |
| `type:feature` | Feature or enhancement |
| `type:process` | Process or policy work |
| `type:workflow` | Workflow automation or guidance |
| `type:guardrail` | Safety gate or guardrail |
| `type:docs` | Documentation work |
| `type:test` | Tests or test infrastructure |
| `type:discussion` | Discussion-only issue |
| `type:umbrella` | Umbrella or tracker issue |
| `type:cleanup` | Cleanup or hygiene work |
* **`observability`**
* Proposed Description: Observability, metrics, and monitoring tasks
* Proposed Color: `5319e7`
* Use: Broader than GlitchTip alone; covers logging, metrics, traces, and general observability pipeline improvements.
## Workflow Status Labels
* **`source:glitchtip`**
* Proposed Description: Issue filed automatically by GlitchTip orchestration
* Proposed Color: `b60205`
* Use: Applied automatically by the orchestrator when a GlitchTip error event is converted into a Gitea issue.
Only one `status:*` label should be active on an issue at a time. When an issue
moves forward, tooling must remove the old `status:*` label and apply the new
one.
* **`status:triage`**
* Proposed Description: Issue needs human or orchestrator triage
* Proposed Color: `fbca04`
* Use: Used for incoming issues (especially automated ones like `source:glitchtip`) that have not yet been evaluated for priority or resolution.
| Label | Use |
| --- | --- |
| `status:triage` | Issue needs triage |
| `status:ready` | Issue is ready for work |
| `status:claimed` | Issue is claimed |
| `status:in-progress` | Issue is being worked on |
| `status:blocked` | Issue is blocked |
| `status:needs-review` | Issue work needs review |
| `status:pr-open` | A linked PR is open |
| `status:approved` | Linked PR is approved |
| `status:merged` | Linked PR is merged |
| `status:reconcile` | Issue needs reconciliation |
| `status:done` | Issue workflow is complete |
| `status:duplicate` | Issue is a duplicate |
| `status:wontfix` | Issue will not be fixed |
## Transition Rules
Suggested lifecycle:
1. New issue created: `status:triage` or `status:ready`
2. Issue selected by an author: `status:claimed`
3. Author starts work: `status:in-progress`
4. Work is blocked: `status:blocked`
5. PR opened: `status:pr-open`
6. PR approved: `status:approved`
7. PR merged but issue still needs closure/reconciliation: `status:reconcile`
8. Issue fully complete: `status:done`
9. Duplicate issue: `status:duplicate`
10. Won't-fix issue: `status:wontfix`
The helper module `issue_workflow_labels.py` is the source of truth for the
canonical label specs and status transition replacement behavior.
## Discussion Issues
Discussion issues must be labeled `type:discussion`.
A discussion issue should not be treated as implementation-ready unless it also
has a clear implementation status and next action.
If a discussion produces implementation work, either:
1. convert the discussion issue into an implementation issue by changing labels
and adding acceptance criteria, or
2. create child implementation issues and leave the discussion issue as
`type:discussion`.
## Tooling
- `manage_labels.py --create-labels` creates the canonical `type:*` and
`status:*` labels.
- `gitea_create_issue` recommends `type:*` and `status:*` labels when missing
and can apply supplied label names.
- `gitea_mark_issue(..., action="start")` replaces old `status:*` labels with
`status:in-progress`.
- `gitea_create_pr` fails closed before PR creation if `status:pr-open` cannot
be applied to the locked issue, then applies it after the PR is created.
- `gitea_set_issue_labels` accepts an explicit `worktree_path` so author
sessions can satisfy the branches-only mutation guard while changing labels.
## Existing Non-Workflow Labels
Existing non-workflow labels such as `mcp`, `workflow`, `labels`, `tracker`,
`jenkins`, `glitchtip`, `documentation`, and `testing` remain valid topical
labels. They do not replace the required `type:*` and `status:*` labels.
docs/llm-workflow-runbooks.md
+10 -5
View File
@@ -581,19 +581,24 @@ loop and do **not** substitute WebFetch/Playwright/manual base64.
- **Profile:** issue-manager or author (any profile allowed to create issues).
- **Steps:** create the parent/roadmap issue; create child issues; apply the
minimal label set; link children to the parent.
- **Labels:** new issues should carry one `type:*` label and one `status:*`
label. Discussion-only issues must carry `type:discussion`. See
[`label-taxonomy.md`](label-taxonomy.md).
- **Prompt:** `Using the issue-manager profile, create issue "<title>" with body
<body>, then create child issues for <list> and link them to the parent.`
### Implement an issue and open a PR
- **Profile:** author.
- **Steps:** claim the issue (`status:in-progress`); create an isolated branch
worktree from latest `master` under `branches/` (`feat/issue-<n>-...` /
- **Steps:** claim the issue (`status:in-progress`, replacing any old
`status:*` label); create an isolated branch worktree from latest `master`
under `branches/` (`feat/issue-<n>-...` /
`fix/...` / `docs/...`); `cd` into that worktree; implement narrowly; add or
update tests if behavior changes; run the full suite; commit with an
issue-linked message; open a PR to `master`. **Do not** review or merge your
own PR. Include an `LLM Handoff Metadata` block (with `LLM-Agent-SHA`) in
the PR body — see [`llm-agent-sha.md`](llm-agent-sha.md).
issue-linked message; open a PR to `master`; move the issue to
`status:pr-open`. **Do not** review or merge your own PR. Include an
`LLM Handoff Metadata` block (with `LLM-Agent-SHA`) in the PR body — see
[`llm-agent-sha.md`](llm-agent-sha.md).
- **Prompt:** `Use an author profile to implement issue #N and open a PR to
master. Do not self-review or self-merge.`
gitea_mcp_server.py
+246 -18
View File
@@ -608,6 +608,7 @@ import already_landed_reconcile # noqa: E402
import author_mutation_worktree # noqa: E402
import issue_claim_heartbeat # noqa: E402
import issue_work_duplicate_gate # noqa: E402
import issue_workflow_labels # noqa: E402
import reviewer_pr_lease # noqa: E402
import merged_cleanup_reconcile # noqa: E402
import reconciler_profile # noqa: E402
@@ -954,6 +955,108 @@ def extract_linked_issue_numbers(text: str | None, branch_name: str | None = Non
issues.update(int(m) for m in pattern.findall(branch_name))
return sorted(list(issues))
def _repo_label_id_map(base: str, auth: str) -> dict[str, int]:
labels = api_get_all(f"{base}/labels", auth) or []
return {
str(lb["name"]): int(lb["id"])
for lb in labels
if isinstance(lb, dict) and lb.get("name") and lb.get("id") is not None
}
def _issue_label_names(base: str, auth: str, issue_number: int) -> list[str]:
issue = api_request("GET", f"{base}/issues/{issue_number}", auth) or {}
return issue_workflow_labels.label_names(issue)
def _put_issue_label_names(
*,
base: str,
auth: str,
issue_number: int,
names: list[str],
label_ids_by_name: dict[str, int] | None = None,
) -> list[dict]:
by_name = label_ids_by_name or _repo_label_id_map(base, auth)
missing = [name for name in names if name not in by_name]
if missing:
raise RuntimeError(
f"The following labels do not exist on the repository: {missing}. "
"Create the canonical workflow labels first."
)
ids = [by_name[name] for name in names]
return api_request(
"PUT",
f"{base}/issues/{issue_number}/labels",
auth,
{"labels": ids},
)
def _transition_issue_status(
*,
base: str,
auth: str,
issue_number: int,
status: str,
label_ids_by_name: dict[str, int] | None = None,
) -> dict:
by_name = label_ids_by_name or _repo_label_id_map(base, auth)
target_status = issue_workflow_labels.canonical_status_label(status)
if target_status not in by_name:
return {
"success": False,
"performed": False,
"issue_number": issue_number,
"target_status": target_status,
"reasons": [
f"required workflow status label '{target_status}' does not exist"
],
}
current = _issue_label_names(base, auth, issue_number)
updated = issue_workflow_labels.transition_status_labels(current, target_status)
_put_issue_label_names(
base=base,
auth=auth,
issue_number=issue_number,
names=updated,
label_ids_by_name=by_name,
)
return {
"success": True,
"performed": True,
"issue_number": issue_number,
"target_status": target_status,
"labels": updated,
}
def _prepare_issue_status_transition(
*,
base: str,
auth: str,
issue_number: int,
status: str,
) -> dict:
by_name = _repo_label_id_map(base, auth)
target_status = issue_workflow_labels.canonical_status_label(status)
if target_status not in by_name:
return {
"success": False,
"performed": False,
"issue_number": issue_number,
"target_status": target_status,
"reasons": [
f"required workflow status label '{target_status}' does not exist"
],
}
current = _issue_label_names(base, auth, issue_number)
updated = issue_workflow_labels.transition_status_labels(current, target_status)
return {
"success": True,
"performed": False,
"issue_number": issue_number,
"target_status": target_status,
"labels": updated,
"label_ids_by_name": by_name,
}
def release_in_progress_label(issue_numbers: list[int], remote: str, host: str | None, org: str | None, repo: str | None) -> dict:
if not issue_numbers:
return {}
@@ -1248,6 +1351,11 @@ def gitea_create_issue(
host: str | None = None,
org: str | None = None,
repo: str | None = None,
labels: list[str] | None = None,
issue_type: str | None = None,
initial_status: str | None = None,
discussion: bool = False,
require_workflow_labels: bool = False,
allow_duplicate_override: bool = False,
split_from_issue: int | None = None,
worktree_path: str | None = None,
@@ -1261,6 +1369,12 @@ def gitea_create_issue(
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
labels: Optional labels to apply by name after creating the issue.
issue_type: Optional canonical type, e.g. 'feature' or 'type:feature'.
initial_status: Optional initial workflow status, e.g. 'ready'.
discussion: If True, require/apply type:discussion.
require_workflow_labels: If True, fail closed unless labels include one
type:* and one status:* label.
allow_duplicate_override: Operator-approved split after duplicate found.
split_from_issue: Existing duplicate issue number when overriding.
worktree_path: Optional path to verify branches-only guard.
@@ -1292,6 +1406,40 @@ def gitea_create_issue(
return blocked
verify_preflight_purity(remote, worktree_path=worktree_path, task="create_issue")
base = repo_api_url(h, o, r)
requested_labels = issue_workflow_labels.labels_for_new_issue(
issue_type=issue_type,
initial_status=initial_status,
extra_labels=labels,
discussion=discussion,
)
label_assessment = issue_workflow_labels.assess_issue_labels(
requested_labels,
discussion=discussion,
require_type=True,
require_status=True,
)
if require_workflow_labels and not label_assessment["valid"]:
return {
"success": False,
"performed": False,
"number": None,
"workflow_label_validation": label_assessment,
"reasons": label_assessment["errors"],
}
label_ids_by_name: dict[str, int] | None = None
if requested_labels:
label_ids_by_name = _repo_label_id_map(base, auth)
missing = [name for name in requested_labels if name not in label_ids_by_name]
if missing:
return {
"success": False,
"performed": False,
"number": None,
"workflow_label_validation": label_assessment,
"reasons": [
f"requested labels do not exist on the repository: {missing}"
],
}
open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth)
closed_issues = api_get_all(
f"{base}/issues?state=closed&type=issues", auth, limit=100
@@ -1326,7 +1474,29 @@ def gitea_create_issue(
_audit("create_issue", host=h, remote=remote, org=o, repo=r,
result=gitea_audit.SUCCEEDED, issue_number=data["number"],
request_metadata={"title": title}, mutation_task="create_issue")
return _with_optional_url({"number": data["number"]}, data.get("html_url"))
result = {
"number": data["number"],
"workflow_label_validation": label_assessment,
}
if requested_labels:
with _audited(
"set_issue_labels",
host=h,
remote=remote,
org=o,
repo=r,
issue_number=data["number"],
request_metadata={"labels": requested_labels, "source": "create_issue"},
):
applied = _put_issue_label_names(
base=base,
auth=auth,
issue_number=data["number"],
names=requested_labels,
label_ids_by_name=label_ids_by_name,
)
result["labels"] = issue_workflow_labels.label_names(applied)
return _with_optional_url(result, data.get("html_url"))
@mcp.tool()
@@ -1735,7 +1905,23 @@ def gitea_create_pr(
)
auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/pulls"
repo_base = repo_api_url(h, o, r)
pr_open_transition = _prepare_issue_status_transition(
base=repo_base,
auth=auth,
issue_number=int(locked_issue),
status="status:pr-open",
)
if not pr_open_transition.get("success"):
return {
"success": False,
"performed": False,
"number": None,
"issue_number": locked_issue,
"issue_status_transition": pr_open_transition,
"reasons": pr_open_transition.get("reasons", []),
}
url = f"{repo_base}/pulls"
payload = {"title": title, "body": body, "head": head, "base": base}
meta = {"title": title, "head": head, "base": base}
try:
@@ -1750,7 +1936,37 @@ def gitea_create_pr(
result=gitea_audit.SUCCEEDED, pr_number=data["number"],
target_branch=head, request_metadata=meta,
mutation_task="create_pr")
return _with_optional_url({"number": data["number"]}, data.get("html_url"))
with _audited(
"set_issue_labels",
host=h,
remote=remote,
org=o,
repo=r,
issue_number=int(locked_issue),
request_metadata={
"source": "create_pr",
"status": "status:pr-open",
"pr_number": data["number"],
},
):
_put_issue_label_names(
base=repo_base,
auth=auth,
issue_number=int(locked_issue),
names=pr_open_transition["labels"],
label_ids_by_name=pr_open_transition["label_ids_by_name"],
)
result = {
"number": data["number"],
"issue_status_transition": {
"success": True,
"performed": True,
"issue_number": int(locked_issue),
"target_status": "status:pr-open",
"labels": pr_open_transition["labels"],
},
}
return _with_optional_url(result, data.get("html_url"))
def _format_list_pr_entry(pr: dict) -> dict:
@@ -6870,15 +7086,8 @@ def gitea_mark_issue(
auth = _auth(h)
base = repo_api_url(h, o, r)
# Find the status:in-progress label id
labels = api_request("GET", f"{base}/labels?limit=100", auth)
label_id = None
for lb in labels:
if lb["name"] == "status:in-progress":
label_id = lb["id"]
break
if label_id is None:
label_ids_by_name = _repo_label_id_map(base, auth)
if "status:in-progress" not in label_ids_by_name:
raise RuntimeError(
"Label 'status:in-progress' not found. "
"Run manage_labels.py to create it first."
@@ -6886,10 +7095,19 @@ def gitea_mark_issue(
if action == "start":
with _audited("label_issue", host=h, remote=remote, org=o, repo=r,
issue_number=issue_number,
request_metadata={"op": "add", "label": "status:in-progress"}):
api_request("POST", f"{base}/issues/{issue_number}/labels", auth,
{"labels": [label_id]})
issue_number=issue_number, request_metadata={
"op": "replace-status",
"label": "status:in-progress",
}):
transition = _transition_issue_status(
base=base,
auth=auth,
issue_number=issue_number,
status="status:in-progress",
label_ids_by_name=label_ids_by_name,
)
if not transition.get("success"):
raise RuntimeError("; ".join(transition.get("reasons") or []))
active_profile = (profile or get_profile().get("profile_name") or "unknown")
branch = (branch_name or "pending").strip() or "pending"
heartbeat_body = issue_claim_heartbeat.format_heartbeat_body(
@@ -6914,8 +7132,10 @@ def gitea_mark_issue(
"message": f"Issue #{issue_number} claimed.",
"heartbeat_posted": heartbeat.get("success", False),
"heartbeat_comment_id": heartbeat.get("comment_id"),
"status_transition": transition,
}
else:
label_id = label_ids_by_name["status:in-progress"]
with _audited("unlabel_issue", host=h, remote=remote, org=o, repo=r,
issue_number=issue_number,
request_metadata={"op": "remove", "label": "status:in-progress"}):
@@ -7281,6 +7501,7 @@ def gitea_create_label(
host: str | None = None,
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""Create a new label on a Gitea repository.
@@ -7292,11 +7513,16 @@ def gitea_create_label(
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
worktree_path: Optional branches worktree to verify before mutation.
Returns:
dict containing the created label details.
"""
verify_preflight_purity(remote)
blocked = _profile_permission_block(
task_capability_map.required_permission("create_label"))
if blocked:
return blocked
verify_preflight_purity(remote, worktree_path=worktree_path, task="create_label")
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)
@@ -7322,6 +7548,7 @@ def gitea_set_issue_labels(
host: str | None = None,
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
) -> list:
"""Replace all labels on a Gitea issue with a new list of label names.
@@ -7332,6 +7559,7 @@ def gitea_set_issue_labels(
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
worktree_path: Optional branches worktree to verify before mutation.
Returns:
list of all labels currently applied to the issue.
@@ -7340,7 +7568,7 @@ def gitea_set_issue_labels(
task_capability_map.required_permission("set_issue_labels"))
if blocked:
return blocked
verify_preflight_purity(remote, task="set_issue_labels")
verify_preflight_purity(remote, worktree_path=worktree_path, task="set_issue_labels")
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)
issue_workflow_labels.py
+194
View File
@@ -0,0 +1,194 @@
"""Canonical issue type/status label taxonomy and transition helpers (#513)."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Iterable, Mapping, Sequence
@dataclass(frozen=True)
class LabelSpec:
name: str
color: str
description: str
TYPE_LABEL_SPECS: tuple[LabelSpec, ...] = (
LabelSpec("type:bug", "b60205", "Bug or defect"),
LabelSpec("type:feature", "a2eeef", "Feature or enhancement"),
LabelSpec("type:process", "5319e7", "Process or policy work"),
LabelSpec("type:workflow", "c2e0c6", "Workflow automation or guidance"),
LabelSpec("type:guardrail", "d93f0b", "Safety gate or guardrail"),
LabelSpec("type:docs", "006b75", "Documentation work"),
LabelSpec("type:test", "0e8a16", "Tests or test infrastructure"),
LabelSpec("type:discussion", "bfdadc", "Discussion-only issue"),
LabelSpec("type:umbrella", "c5def5", "Umbrella or tracker issue"),
LabelSpec("type:cleanup", "ededed", "Cleanup or hygiene work"),
)
STATUS_LABEL_SPECS: tuple[LabelSpec, ...] = (
LabelSpec("status:triage", "fbca04", "Issue needs triage"),
LabelSpec("status:ready", "0e8a16", "Issue is ready for work"),
LabelSpec("status:claimed", "fefe2e", "Issue is claimed"),
LabelSpec("status:in-progress", "fefe2e", "Issue is being worked on"),
LabelSpec("status:blocked", "b60205", "Issue is blocked"),
LabelSpec("status:needs-review", "0052cc", "Issue work needs review"),
LabelSpec("status:pr-open", "1d76db", "A linked PR is open"),
LabelSpec("status:approved", "0e8a16", "Linked PR is approved"),
LabelSpec("status:merged", "5319e7", "Linked PR is merged"),
LabelSpec("status:reconcile", "d93f0b", "Issue needs reconciliation"),
LabelSpec("status:done", "0e8a16", "Issue workflow is complete"),
LabelSpec("status:duplicate", "cccccc", "Issue is a duplicate"),
LabelSpec("status:wontfix", "000000", "Issue will not be fixed"),
)
CANONICAL_LABEL_SPECS: tuple[LabelSpec, ...] = (
TYPE_LABEL_SPECS + STATUS_LABEL_SPECS
)
TYPE_LABELS: frozenset[str] = frozenset(spec.name for spec in TYPE_LABEL_SPECS)
STATUS_LABELS: frozenset[str] = frozenset(spec.name for spec in STATUS_LABEL_SPECS)
CANONICAL_LABELS: frozenset[str] = frozenset(
spec.name for spec in CANONICAL_LABEL_SPECS
)
STATUS_TRANSITIONS: dict[str, str] = {
"triage": "status:triage",
"ready": "status:ready",
"claim": "status:claimed",
"claimed": "status:claimed",
"start": "status:in-progress",
"start_work": "status:in-progress",
"in_progress": "status:in-progress",
"in-progress": "status:in-progress",
"block": "status:blocked",
"blocked": "status:blocked",
"needs_review": "status:needs-review",
"needs-review": "status:needs-review",
"pr_open": "status:pr-open",
"pr-open": "status:pr-open",
"approved": "status:approved",
"merge": "status:reconcile",
"merged": "status:reconcile",
"reconcile": "status:reconcile",
"done": "status:done",
"complete": "status:done",
"duplicate": "status:duplicate",
"wontfix": "status:wontfix",
}
def label_name(label: str | Mapping[str, object]) -> str:
"""Return a normalized label name from a Gitea label object or string."""
if isinstance(label, str):
return label.strip()
value = label.get("name")
return str(value or "").strip()
def label_names(labels: Iterable[str | Mapping[str, object]] | Mapping[str, object]) -> list[str]:
"""Extract label names from a label list or issue-like mapping."""
raw: object
if isinstance(labels, Mapping):
raw = labels.get("labels", [])
else:
raw = labels
return [name for name in (label_name(label) for label in raw or []) if name]
def type_labels(labels: Iterable[str | Mapping[str, object]] | Mapping[str, object]) -> list[str]:
return [name for name in label_names(labels) if name.startswith("type:")]
def status_labels(labels: Iterable[str | Mapping[str, object]] | Mapping[str, object]) -> list[str]:
return [name for name in label_names(labels) if name.startswith("status:")]
def canonical_status_label(status_or_transition: str) -> str:
status = status_or_transition.strip()
if status in STATUS_LABELS:
return status
normalized = status.lower().replace(" ", "-")
try:
return STATUS_TRANSITIONS[normalized]
except KeyError as exc:
raise ValueError(
f"unknown workflow status or transition '{status_or_transition}'"
) from exc
def transition_status_labels(
existing_labels: Iterable[str | Mapping[str, object]] | Mapping[str, object],
status_or_transition: str,
) -> list[str]:
"""Replace all active status labels with the requested canonical status."""
new_status = canonical_status_label(status_or_transition)
kept = [name for name in label_names(existing_labels) if not name.startswith("status:")]
if new_status not in kept:
kept.append(new_status)
return kept
def labels_for_new_issue(
issue_type: str | None = None,
initial_status: str | None = None,
extra_labels: Sequence[str] | None = None,
*,
discussion: bool = False,
) -> list[str]:
names = list(extra_labels or [])
if issue_type:
type_name = issue_type if issue_type.startswith("type:") else f"type:{issue_type}"
names.append(type_name)
if discussion:
names.append("type:discussion")
if initial_status:
names.append(canonical_status_label(initial_status))
result: list[str] = []
for name in names:
if name not in result:
result.append(name)
return result
def assess_issue_labels(
labels: Iterable[str | Mapping[str, object]] | Mapping[str, object],
*,
discussion: bool = False,
require_type: bool = True,
require_status: bool = True,
) -> dict:
names = label_names(labels)
found_types = type_labels(names)
found_statuses = status_labels(names)
errors: list[str] = []
warnings: list[str] = []
if require_type and not found_types:
errors.append("issue is missing a type:* label")
if require_status and not found_statuses:
errors.append("issue is missing a status:* label")
if discussion and "type:discussion" not in found_types:
errors.append("discussion issue is missing type:discussion")
if len(found_statuses) > 1:
errors.append(
"issue has multiple active status:* labels: "
+ ", ".join(found_statuses)
)
for name in found_types:
if name not in TYPE_LABELS:
warnings.append(f"unknown type label '{name}'")
for name in found_statuses:
if name not in STATUS_LABELS:
warnings.append(f"unknown status label '{name}'")
return {
"valid": not errors,
"labels": names,
"type_labels": found_types,
"status_labels": found_statuses,
"errors": errors,
"warnings": warnings,
}
manage_labels.py
+5 -2
View File
@@ -23,6 +23,7 @@ if os.path.exists(venv_python) and sys.executable != venv_python:
os.execv(venv_python, [venv_python] + sys.argv)
from gitea_auth import get_auth_header, api_request, repo_api_url
import issue_workflow_labels
HOST = "gitea.dadeschools.net"
ORG = "Contractor"
@@ -35,8 +36,10 @@ LABELS = [
{"name": "epic", "color": "8250df", "description": ""},
{"name": "important", "color": "fbca04", "description": ""},
{"name": "nice-to-have", "color": "0e8a16", "description": ""},
{"name": "status:in-progress", "color": "fefe2e",
"description": "Issue is being worked on"},
*[
{"name": spec.name, "color": spec.color, "description": spec.description}
for spec in issue_workflow_labels.CANONICAL_LABEL_SPECS
],
]
# issue number -> label names to apply (one-off backfill)
task_capability_map.py
+5
View File
@@ -36,6 +36,10 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.issue.comment",
"role": "author",
},
"create_label": {
"permission": "gitea.issue.comment",
"role": "author",
},
"create_branch": {
"permission": "gitea.branch.create",
"role": "author",
@@ -155,6 +159,7 @@ ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
"gitea_create_issue_comment": "comment_issue",
"gitea_mark_issue": "mark_issue",
"gitea_set_issue_labels": "set_issue_labels",
"gitea_create_label": "create_label",
"gitea_commit_files": "commit_files",
}
tests/test_create_issue.py
+31
View File
@@ -103,6 +103,37 @@ class TestAPIPayload(unittest.TestCase):
self.assertEqual(payload["title"], "My Title")
self.assertEqual(payload["body"], "My Body")
@patch("create_issue.get_credentials", return_value=FAKE_CREDS)
def test_applies_workflow_labels_by_name(self, _cred):
def api_side_effect(method, url, auth, payload=None):
if method == "GET" and url.endswith("/labels"):
return [
{"id": 1, "name": "type:feature"},
{"id": 2, "name": "status:ready"},
]
if method == "POST" and url.endswith("/issues"):
return {"number": 7, "html_url": "http://x/7"}
if method == "PUT" and url.endswith("/issues/7/labels"):
return [{"name": "type:feature"}, {"name": "status:ready"}]
return {}
with patch("create_issue.api_request", side_effect=api_side_effect) as mock_api:
rc = create_issue.main([
"--title", "My Title",
"--type-label", "feature",
"--status-label", "ready",
])
self.assertEqual(rc, 0)
put_call = next(c for c in mock_api.call_args_list if c[0][0] == "PUT")
self.assertEqual(put_call[0][3], {"labels": [1, 2]})
@patch("create_issue.get_credentials", return_value=FAKE_CREDS)
def test_require_workflow_labels_blocks_missing_labels(self, _cred):
with patch("create_issue.api_request") as mock_api:
rc = create_issue.main(["--title", "My Title", "--require-workflow-labels"])
self.assertEqual(rc, 1)
mock_api.assert_not_called()
@patch("create_issue.get_credentials", return_value=FAKE_CREDS)
def test_url_construction(self, _cred):
with patch("create_issue.api_request",
tests/test_issue_workflow_labels.py
+73
View File
@@ -0,0 +1,73 @@
"""Tests for canonical issue workflow label policy (#513)."""
from __future__ import annotations
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import issue_workflow_labels as labels # noqa: E402
class TestIssueWorkflowLabelValidation(unittest.TestCase):
def test_valid_labeled_issue(self):
result = labels.assess_issue_labels(["type:feature", "status:ready"])
self.assertTrue(result["valid"])
self.assertEqual(result["type_labels"], ["type:feature"])
self.assertEqual(result["status_labels"], ["status:ready"])
def test_issue_missing_type_label(self):
result = labels.assess_issue_labels(["status:ready"])
self.assertFalse(result["valid"])
self.assertIn("issue is missing a type:* label", result["errors"])
def test_issue_missing_status_label(self):
result = labels.assess_issue_labels(["type:bug"])
self.assertFalse(result["valid"])
self.assertIn("issue is missing a status:* label", result["errors"])
def test_discussion_issue_missing_type_discussion(self):
result = labels.assess_issue_labels(
["type:feature", "status:triage"],
discussion=True,
)
self.assertFalse(result["valid"])
self.assertIn("discussion issue is missing type:discussion", result["errors"])
def test_multiple_conflicting_status_labels(self):
result = labels.assess_issue_labels(
["type:feature", "status:ready", "status:blocked"]
)
self.assertFalse(result["valid"])
self.assertTrue(
any("multiple active status:* labels" in err for err in result["errors"])
)
class TestIssueWorkflowStatusTransitions(unittest.TestCase):
def test_status_transition_removes_old_status_label(self):
result = labels.transition_status_labels(
["type:feature", "status:ready", "workflow"],
"in-progress",
)
self.assertEqual(result, ["type:feature", "workflow", "status:in-progress"])
def test_pr_open_transition_applies_status_pr_open(self):
result = labels.transition_status_labels(
["type:feature", "status:in-progress"],
"pr-open",
)
self.assertEqual(result, ["type:feature", "status:pr-open"])
def test_duplicate_transition_applies_status_duplicate(self):
result = labels.transition_status_labels(
["type:process", "status:triage"],
"duplicate",
)
self.assertEqual(result, ["type:process", "status:duplicate"])
if __name__ == "__main__":
unittest.main()
tests/test_issue_write_tool_gates.py
+6 -5
View File
@@ -164,13 +164,14 @@ class TestAllowedIssueWritesProceed(IssueWriteGateBase):
result = mcp_server.gitea_close_issue(issue_number=9, remote="prgs")
self.assertTrue(result.get("success"))
@patch("mcp_server.api_get_all", return_value=[{"id": 10, "name": "status:in-progress"}])
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_resolver_allowed_mark_issue_proceeds(self, _auth, mock_api):
def test_resolver_allowed_mark_issue_proceeds(self, _auth, mock_api, _labels):
def api_side_effect(method, url, auth, payload=None):
if method == "GET" and url.endswith("/labels?limit=100"):
return [{"id": 10, "name": "status:in-progress"}]
if method == "POST" and "/issues/9/labels" in url:
if method == "GET" and "/issues/9" in url:
return {"labels": []}
if method == "PUT" and "/issues/9/labels" in url:
return [{"name": "status:in-progress"}]
if method == "POST" and "/issues/9/comments" in url:
return {"id": 501}
@@ -209,4 +210,4 @@ class TestCreateIssueMissingPermissionReport(IssueWriteGateBase):
if __name__ == "__main__":
unittest.main()
unittest.main()
tests/test_manage_labels.py
+6
View File
@@ -11,6 +11,7 @@ from unittest.mock import MagicMock, call, patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import manage_labels # noqa: E402
import issue_workflow_labels # noqa: E402
FAKE_AUTH = "Basic dGVzdDp0ZXN0" # base64("test:test")
@@ -129,6 +130,11 @@ class TestConstants(unittest.TestCase):
names = [l["name"] for l in manage_labels.LABELS]
self.assertIn("status:in-progress", names)
def test_canonical_workflow_labels_are_defined(self):
names = {l["name"] for l in manage_labels.LABELS}
for spec in issue_workflow_labels.CANONICAL_LABEL_SPECS:
self.assertIn(spec.name, names)
def test_all_mapped_labels_are_defined(self):
defined = {l["name"] for l in manage_labels.LABELS}
for issue, names in manage_labels.MAPPING.items():
tests/test_mcp_server.py
+121 -18
View File
@@ -85,6 +85,16 @@ def _visible_approval_reviews(reviewer="reviewer-bot", sha="abc123"):
_DEFAULT_LEASE_SESSION = "mcp-test-reviewer-lease"
_NO_PR_WORK_LEASE_BLOCK = {"block": False, "reasons": [], "mutation_allowed": True}
WORKFLOW_REPO_LABELS = [
{"id": 1, "name": "type:feature"},
{"id": 2, "name": "status:in-progress"},
{"id": 3, "name": "status:ready"},
{"id": 4, "name": "status:pr-open"},
]
def _issue_with_labels(*names):
return {"labels": [{"name": name} for name in names]}
def _reviewer_lease_comment(
@@ -293,6 +303,22 @@ class TestCreateIssue(unittest.TestCase):
self.assertIn("gitea.prgs.cc", url)
self.assertIn("Scaled-Tech-Consulting", url)
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_issue_required_workflow_labels_blocks(self, _auth, _get_all, mock_api, _role):
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
result = gitea_create_issue(
title="Test issue",
require_workflow_labels=True,
)
self.assertFalse(result["success"])
self.assertFalse(result["performed"])
self.assertIn("issue is missing a type:* label", result["reasons"])
mock_api.assert_not_called()
# ---------------------------------------------------------------------------
# Create PR
@@ -303,13 +329,24 @@ class TestCreatePR(unittest.TestCase):
"mcp_server.issue_duplicate_context_fetcher",
return_value=([], [], {"status": "not_claimed"}),
)
@patch("mcp_server.api_get_all", return_value=WORKFLOW_REPO_LABELS)
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_creates_pr(self, _auth, mock_api, _role, _dup_fetcher):
def test_creates_pr(self, _auth, mock_api, _role, _labels, _dup_fetcher):
worktree = os.path.realpath(os.getcwd())
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
def api_side_effect(method, url, auth, payload=None):
if method == "GET" and "/issues/123" in url:
return _issue_with_labels("type:feature", "status:in-progress")
if method == "POST" and url.endswith("/pulls"):
return {"number": 3, "html_url": "https://example.com/pulls/3"}
if method == "PUT" and url.endswith("/issues/123/labels"):
return [{"name": "type:feature"}, {"name": "status:pr-open"}]
return {}
mock_api.side_effect = api_side_effect
with tempfile.TemporaryDirectory() as lock_dir:
env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir}
with patch.dict(os.environ, env, clear=True):
@@ -322,22 +359,37 @@ class TestCreatePR(unittest.TestCase):
)
self.assertEqual(result["number"], 3)
self.assertNotIn("url", result)
payload = mock_api.call_args[0][3]
post_call = next(c for c in mock_api.call_args_list if c[0][0] == "POST")
payload = post_call[0][3]
self.assertEqual(payload["head"], "feat/x")
self.assertEqual(payload["base"], "main")
self.assertIn("Closes #123", payload["title"])
put_call = next(c for c in mock_api.call_args_list if c[0][0] == "PUT")
self.assertEqual(put_call[0][3], {"labels": [1, 4]})
self.assertTrue(result["issue_status_transition"]["performed"])
@patch(
"mcp_server.issue_duplicate_context_fetcher",
return_value=([], [], {"status": "not_claimed"}),
)
@patch("mcp_server.api_get_all", return_value=WORKFLOW_REPO_LABELS)
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_reveal_opt_in_includes_url(self, _auth, mock_api, _role, _dup_fetcher):
def test_create_pr_reveal_opt_in_includes_url(self, _auth, mock_api, _role, _labels, _dup_fetcher):
worktree = os.path.realpath(os.getcwd())
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
def api_side_effect(method, url, auth, payload=None):
if method == "GET" and "/issues/123" in url:
return _issue_with_labels("type:feature", "status:in-progress")
if method == "POST" and url.endswith("/pulls"):
return {"number": 3, "html_url": "https://example.com/pulls/3"}
if method == "PUT" and url.endswith("/issues/123/labels"):
return [{"name": "type:feature"}, {"name": "status:pr-open"}]
return {}
mock_api.side_effect = api_side_effect
with tempfile.TemporaryDirectory() as lock_dir:
env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
with patch.dict(os.environ, env, clear=True):
@@ -350,6 +402,38 @@ class TestCreatePR(unittest.TestCase):
)
self.assertIn("pulls/3", result["url"])
@patch(
"mcp_server.issue_duplicate_context_fetcher",
return_value=([], [], {"status": "not_claimed"}),
)
@patch("mcp_server.api_get_all", return_value=[{"id": 1, "name": "type:feature"}])
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_blocks_when_pr_open_status_label_missing(
self, _auth, mock_api, _role, _labels, _dup_fetcher
):
worktree = os.path.realpath(os.getcwd())
mock_api.return_value = _issue_with_labels("type:feature", "status:in-progress")
with tempfile.TemporaryDirectory() as lock_dir:
env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir}
with patch.dict(os.environ, env, clear=True):
_bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree)
result = gitea_create_pr(
title="feat: X Closes #123",
head="feat/x",
base="main",
worktree_path=worktree,
)
self.assertFalse(result["success"])
self.assertFalse(result["performed"])
self.assertIn("status:pr-open", result["reasons"][0])
self.assertFalse(
any(c[0][0] == "POST" and c[0][1].endswith("/pulls")
for c in mock_api.call_args_list)
)
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@@ -474,26 +558,33 @@ class TestViewIssue(unittest.TestCase):
# ---------------------------------------------------------------------------
class TestMarkIssue(unittest.TestCase):
@patch("mcp_server.api_get_all", return_value=WORKFLOW_REPO_LABELS)
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_start_adds_label(self, _auth, mock_api):
# labels, add label, post claim heartbeat comment
mock_api.side_effect = [
[{"id": 10, "name": "status:in-progress"}],
[{"name": "status:in-progress"}],
{"id": 99},
]
def test_start_adds_label(self, _auth, mock_api, _labels):
def api_side_effect(method, url, auth, payload=None):
if method == "GET" and "/issues/5" in url:
return _issue_with_labels("type:feature", "status:ready")
if method == "PUT" and url.endswith("/issues/5/labels"):
return [{"name": "type:feature"}, {"name": "status:in-progress"}]
if method == "POST" and url.endswith("/issues/5/comments"):
return {"id": 99}
return {}
mock_api.side_effect = api_side_effect
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
result = gitea_mark_issue(issue_number=5, action="start")
self.assertTrue(result["success"])
self.assertIn("claimed", result["message"])
self.assertTrue(result.get("heartbeat_posted"))
put_call = next(c for c in mock_api.call_args_list if c[0][0] == "PUT")
self.assertEqual(put_call[0][3], {"labels": [1, 2]})
@patch("mcp_server.api_get_all", return_value=WORKFLOW_REPO_LABELS)
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_done_removes_label(self, _auth, mock_api):
def test_done_removes_label(self, _auth, mock_api, _labels):
mock_api.side_effect = [
[{"id": 10, "name": "status:in-progress"}],
None,
]
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
@@ -505,10 +596,10 @@ class TestMarkIssue(unittest.TestCase):
with self.assertRaises(ValueError):
gitea_mark_issue(issue_number=5, action="pause")
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_missing_label_raises(self, _auth, mock_api):
mock_api.return_value = [] # no labels exist
def test_missing_label_raises(self, _auth, mock_api, _labels):
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
with self.assertRaises(RuntimeError):
gitea_mark_issue(issue_number=5, action="start")
@@ -3750,13 +3841,24 @@ class TestIssueLocking(unittest.TestCase):
)
self.assertIn("lock provenance", str(ctx.exception).lower())
@patch("mcp_server.api_get_all", return_value=WORKFLOW_REPO_LABELS)
@patch("mcp_server.api_request")
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_honors_scratch_worktree_lock(self, _auth, _role, mock_api):
def test_create_pr_honors_scratch_worktree_lock(self, _auth, _role, mock_api, _labels):
scratch = os.path.realpath("/tmp/gitea-tools-author-scratch/issue-249-e2e")
mock_api.return_value = {"number": 250, "html_url": "https://example/pr/250"}
def api_side_effect(method, url, auth, payload=None):
if method == "GET" and "/issues/249" in url:
return _issue_with_labels("type:feature", "status:in-progress")
if method == "POST" and url.endswith("/pulls"):
return {"number": 250, "html_url": "https://example/pr/250"}
if method == "PUT" and url.endswith("/issues/249/labels"):
return [{"name": "type:feature"}, {"name": "status:pr-open"}]
return {}
mock_api.side_effect = api_side_effect
_bind_test_lock(
issue_number=249,
branch_name="feat/issue-249-issue-lock-scratch-worktree",
@@ -3771,6 +3873,7 @@ class TestIssueLocking(unittest.TestCase):
worktree_path=scratch,
)
self.assertEqual(res["number"], 250)
self.assertTrue(res["issue_status_transition"]["performed"])
# ---------------------------------------------------------------------------