gitea-admin/cowork-local
2
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

create feature mcp-issue-context
CI / test (pull_request) Canceled after 0s
Details

Browse Source
This commit is contained in:
taiph9
2026-08-25 22:54:56 +09:00
parent 86c27e2e79
commit 98f4a1ed77
2 changed files with 1018 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mcp_servers/project_context/providers/issue.py
+372 -2
View File
@@ -1,11 +1,124 @@
"""Provider boundary owned with get_project_issue_context."""
"""Provider boundary owned with get_project_issue_context.
Read-only adapter that maps a Gitea issue/pull-request onto the neutral
``IssueContextOutput`` schema declared in ``tools/issue_context.py``. Nothing
here changes the shared runtime, the registry, or the tool schema — this
module only fills in ``build_provider`` so ``runtime.py``'s
``PROVIDER_FACTORIES["get_project_issue_context"]`` resolves to a real,
read-only Gitea client instead of :class:`UnconfiguredIssueProvider`.
Design notes (documented here so a reviewer does not have to guess):
* Credentials (``GITEA_TOKEN``) and the base URL are read INSIDE
``build_provider``/``_resolve_target`` only — never at import time — because
``build_provider`` is only ever invoked by
``ProjectProviderResolver.resolve`` (see ``runtime.py``), which
``server.py::dispatch`` calls strictly AFTER ``policy.decide`` returns
``True``. Reading env vars at module scope would read them before the
policy check ever runs.
* The provider is bound to exactly one project/repository at construction
time (``identity.project`` -> ``PROJECT_CONTEXT_REPO_MAP``). The
``project_id`` argument received per call is only used to echo it back in
the response and as a defensive equality check — it is never used to pick
which repository to query. This means a caller can never redirect the
provider to an arbitrary repository by tampering with ``project_id``.
* ``revision`` pins to the issue's own ``updated_at`` timestamp. Gitea issues
(unlike commits/PRs) have no natural commit SHA of their own, so the most
meaningful, verifiable "version marker" Gitea offers for an issue is its
last-modified time: given the same URL and the same ``revision`` string, a
reviewer can confirm whether the issue has changed since this was fetched.
* ``related`` items are derived from BARE ``#<number>`` cross-references
found in the issue body — never from a ``#<number>`` that is only the
label text of an explicit Markdown link (``[other-repo PR #4](http://.../
other-repo/pulls/4)``). A real issue body regularly links to a pull
request or an issue in a DIFFERENT repository this way; re-guessing that
as "issue #4 in THIS repo" would silently point at the wrong resource
entirely, which is worse than omitting it — the author already gave the
correct URL right there, so extraction strips whole ``[label](url)``
spans, then any remaining bare URLs, before scanning for mentions. Titles
for the mentions that DO survive are NOT fetched with an extra API call
(kept deliberately simple and bounded) — each related item's title is a
generic, honest placeholder ("Referenced item #<n>"), and its ``url``
always resolves to a real, openable page in this same repository. This
is a known, documented simplification, not a bug.
* ``acceptance_criteria`` looks for the issue's own "Acceptance Criteria"
Markdown heading (``#`` through ``######``, case-insensitive) and only
collects checklist items (``- [ ]``/``- [x]``) from inside that section,
up to the next heading of equal-or-shallower depth. This matters because
a real work-item body commonly has a SEPARATE "Definition of Done"
section that also uses checklist syntax — without scoping to the
matching heading, those unrelated items would be silently folded into
"acceptance criteria" and misrepresent the issue. When no such heading
exists at all (ad-hoc issues with no fixed template), this falls back to
scanning the whole body, so those issues still get a best-effort result
instead of an always-empty list.
* ``truncated``/``returned``/``remaining``/``next_cursor`` describe pagination
over the ``related`` list only. ``description`` has its own independent,
generous hard safety cap (so the response stays bounded even for a
pathological issue body) and is never silently cut without a visible
notice appended.
* The regexes that derive ``acceptance_criteria``/``related`` never scan more
than ``_MAX_SCAN_CHARS`` of the raw body, and always run on a copy with
``http(s)://`` links stripped first — otherwise a `#42` inside a URL
fragment (e.g. a doc anchor) would be mis-detected as a cross-reference to
issue #42, and a pathologically large body would cost unbounded regex work
on every call regardless of the display cap on ``description``.
"""
from __future__ import annotations
import json
import os
import re
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Protocol
import requests
from ..foundation import IdentityContext, ProviderError
# ---- tunables (documented, not hardcoded secrets) -------------------------
_REQUEST_TIMEOUT_SECONDS = 10
_STANDARD_RELATED_PAGE_SIZE = 20
_FULL_RELATED_PAGE_SIZE = 100
_SUMMARY_DESCRIPTION_CHARS = 280
_MAX_DESCRIPTION_CHARS = 20_000
_MAX_SCAN_CHARS = 200_000 # hard cap on regex work, independent of the display cap above
_TRUNCATION_NOTICE = "\n\n[description truncated: exceeds the display size limit]"
_ISSUE_KEY_PATTERN = re.compile(r"^[1-9][0-9]*$")
_CHECKLIST_PATTERN = re.compile(r"^[-*]\s+\[[ xX]\]\s+(.+)$", re.MULTILINE)
_MENTION_PATTERN = re.compile(r"(?<!\w)#([1-9][0-9]*)\b")
_URL_PATTERN = re.compile(r"https?://\S+")
# A whole Markdown link span, label + target together — stripped as ONE unit
# so a `#<number>` that is only the link's label text (often a cross-repo or
# pull-request reference) is never re-guessed as a same-repo issue mention.
_MARKDOWN_LINK_PATTERN = re.compile(r"\[[^\]]*\]\([^)]*\)")
# ATX heading line, e.g. "# Acceptance Criteria" / "## Acceptance Criteria".
_HEADING_PATTERN = re.compile(r"^(#{1,6})[ \t]+(.+?)\s*$", re.MULTILINE)
_ACCEPTANCE_HEADING_NAMES = ("acceptance criteria",)
def _extract_heading_section(text: str, heading_names: tuple[str, ...]) -> str | None:
"""Return the body of the first ATX heading whose title case-insensitively
matches one of ``heading_names``, up to the next heading of equal or
shallower depth (or the end of ``text``). Returns ``None`` when no such
heading exists, so the caller can fall back to the whole body."""
wanted = {name.strip().lower() for name in heading_names}
headings = list(_HEADING_PATTERN.finditer(text))
for index, match in enumerate(headings):
if match.group(2).strip().lower() not in wanted:
continue
level = len(match.group(1))
end = len(text)
for later in headings[index + 1 :]:
if len(later.group(1)) <= level:
end = later.start()
break
return text[match.end() : end]
return None
class IssueProvider(Protocol):
def get_issue_context(self, **arguments: Any) -> dict[str, Any]: ...
@@ -20,6 +133,263 @@ class UnconfiguredIssueProvider:
)
@dataclass(frozen=True)
class _GiteaRepoTarget:
base_url: str
owner: str
repo: str
token: str
project_id: str
def _load_repo_map() -> dict[str, str]:
raw = os.environ.get("PROJECT_CONTEXT_REPO_MAP", "").strip()
if not raw:
return {}
try:
parsed = json.loads(raw)
except json.JSONDecodeError as exc:
raise ProviderError(
"UNAVAILABLE",
"PROJECT_CONTEXT_REPO_MAP is not valid JSON.",
retryable=False,
) from exc
if not isinstance(parsed, dict) or not all(
isinstance(k, str) and isinstance(v, str) for k, v in parsed.items()
):
raise ProviderError(
"UNAVAILABLE",
"PROJECT_CONTEXT_REPO_MAP must be a JSON object of project_id -> 'owner/repo'.",
retryable=False,
)
return parsed
def _resolve_target(identity: IdentityContext) -> _GiteaRepoTarget:
"""Read-only credential/target resolution. Only ever called AFTER the
runtime's policy has already granted ALLOW for this identity/tool."""
base_url = os.environ.get("GITEA_BASE_URL", "").strip().rstrip("/")
token = os.environ.get("GITEA_TOKEN", "").strip()
if not base_url or not token:
raise ProviderError(
"UNAVAILABLE",
"GITEA_BASE_URL/GITEA_TOKEN are not configured for this environment.",
retryable=False,
)
repo_map = _load_repo_map()
slug = repo_map.get(identity.project, "")
if not slug or "/" not in slug:
raise ProviderError(
"UNAVAILABLE",
"This project is not mapped to an approved Gitea repository.",
retryable=False,
)
owner, _, repo = slug.partition("/")
if not owner or not repo:
raise ProviderError(
"UNAVAILABLE",
"This project's Gitea repository mapping is malformed.",
retryable=False,
)
return _GiteaRepoTarget(
base_url=base_url, owner=owner, repo=repo, token=token, project_id=identity.project,
)
def build_provider(identity: IdentityContext) -> IssueProvider:
"""Replace only this factory when wiring the approved read-only issue adapter."""
return UnconfiguredIssueProvider()
return GiteaIssueProvider(_resolve_target(identity))
class GiteaIssueProvider:
"""Read-only adapter mapping one Gitea issue/PR onto the neutral schema."""
def __init__(self, target: _GiteaRepoTarget) -> None:
self._target = target
def get_issue_context(
self,
*,
project_id: str,
issue_key: str,
detail: str,
cursor: str | None,
**_: Any,
) -> dict[str, Any]:
if project_id != self._target.project_id:
# Defense in depth: the runtime's policy already guarantees this
# can never happen (DENIED would have fired first), but the
# provider never trusts caller-supplied routing regardless.
raise ProviderError(
"INTERNAL",
"Resolved provider does not match the requested project.",
retryable=False,
)
if not _ISSUE_KEY_PATTERN.match(issue_key):
raise ProviderError(
"INVALID_INPUT",
"issue_key must be a positive work item number.",
retryable=False,
)
offset = self._decode_cursor(cursor)
payload = self._fetch_issue(issue_key)
title = str(payload.get("title") or "")
raw_state = str(payload.get("state") or "")
status = raw_state if raw_state in {"open", "closed"} else "unknown"
body = str(payload.get("body") or "")
description = self._build_description(body, detail)
# Bounded regardless of the actual body size: caps worst-case regex
# cost, independently of `description`'s own display-only cap.
scan_text = body[:_MAX_SCAN_CHARS]
acceptance_section = _extract_heading_section(scan_text, _ACCEPTANCE_HEADING_NAMES)
acceptance_criteria = tuple(
_CHECKLIST_PATTERN.findall(acceptance_section if acceptance_section is not None else scan_text)
)
related_all = self._extract_related(scan_text, issue_key)
related_page, returned, remaining, truncated, next_cursor = self._paginate_related(
related_all, detail, offset,
)
html_url = str(
payload.get("html_url")
or f"{self._target.base_url}/{self._target.owner}/{self._target.repo}/issues/{issue_key}"
)
updated_at = str(payload.get("updated_at") or "")
retrieved_at = datetime.now(timezone.utc).isoformat()
return {
"project_id": project_id,
"issue_key": issue_key,
"title": title,
"status": status,
"description": description,
"acceptance_criteria": acceptance_criteria,
"related": related_page,
"source": {
"system": "gitea",
"url": html_url,
"revision": f"issue-updated:{updated_at or retrieved_at}",
"retrieved_at": retrieved_at,
},
"truncated": truncated,
"returned": returned,
"remaining": remaining,
"next_cursor": next_cursor,
}
# ---- internals ---------------------------------------------------
def _decode_cursor(self, cursor: str | None) -> int:
if cursor is None:
return 0
try:
offset = int(cursor)
except ValueError as exc:
raise ProviderError("INVALID_INPUT", "cursor is not valid.", retryable=False) from exc
if offset < 0:
raise ProviderError("INVALID_INPUT", "cursor is not valid.", retryable=False)
return offset
def _build_description(self, body: str, detail: str) -> str:
text = body.strip()
if detail == "summary":
return text.split("\n\n", 1)[0][:_SUMMARY_DESCRIPTION_CHARS]
if len(text) > _MAX_DESCRIPTION_CHARS:
return text[:_MAX_DESCRIPTION_CHARS] + _TRUNCATION_NOTICE
return text
def _extract_related(self, body: str, issue_key: str) -> tuple[dict[str, str], ...]:
# Strip whole `[label](url)` spans FIRST (as one unit) so a `#<number>`
# that only appears as a Markdown link's label — often a cross-repo or
# pull-request reference with its own, possibly different, URL right
# there — is never re-guessed as "issue #<number> in this repo".
text_without_links = _MARKDOWN_LINK_PATTERN.sub(" ", body)
# Then strip any remaining bare URLs so a doc-anchor link like
# ".../guide#42" is never mistaken for a cross-reference to issue #42.
text_without_urls = _URL_PATTERN.sub(" ", text_without_links)
numbers = sorted({int(n) for n in _MENTION_PATTERN.findall(text_without_urls) if n != issue_key})
return tuple(
{
"item_id": str(number),
"relation": "mentioned",
"title": f"Referenced item #{number}",
"url": f"{self._target.base_url}/{self._target.owner}/{self._target.repo}/issues/{number}",
}
for number in numbers
)
def _paginate_related(
self,
related_all: tuple[dict[str, str], ...],
detail: str,
offset: int,
) -> tuple[tuple[dict[str, str], ...], int, int, bool, str | None]:
if detail == "summary":
# Summary mode intentionally omits related items outright; it is
# not a size-limit truncation, so callers who need them must
# call again with detail="standard"/"full".
remaining = len(related_all)
return (), 0, remaining, remaining > 0, None
page_size = _FULL_RELATED_PAGE_SIZE if detail == "full" else _STANDARD_RELATED_PAGE_SIZE
page = related_all[offset : offset + page_size]
remaining = max(0, len(related_all) - (offset + page_size))
truncated = remaining > 0
next_cursor = str(offset + page_size) if truncated else None
return page, len(page), remaining, truncated, next_cursor
def _fetch_issue(self, issue_key: str) -> dict[str, Any]:
url = (
f"{self._target.base_url}/api/v1/repos/{self._target.owner}/"
f"{self._target.repo}/issues/{issue_key}"
)
headers = {"Authorization": f"token {self._target.token}"}
try:
response = requests.get(url, headers=headers, timeout=_REQUEST_TIMEOUT_SECONDS)
except requests.exceptions.Timeout as exc:
raise ProviderError(
"UPSTREAM_TIMEOUT", "The Gitea request timed out.", retryable=True,
) from exc
except requests.exceptions.RequestException as exc:
# Never surface str(exc) — it can embed the request URL/host and,
# in some transport errors, request headers.
raise ProviderError(
"UPSTREAM_ERROR", "The Gitea request failed.", retryable=True,
) from exc
if response.status_code == 404:
raise ProviderError(
"NOT_FOUND",
"The work item was not found or is not accessible.",
retryable=False,
)
if response.status_code == 429:
raise ProviderError("RATE_LIMITED", "Gitea rate-limited this request.", retryable=True)
if response.status_code in (401, 403):
raise ProviderError(
"UPSTREAM_ERROR",
"The read-only Gitea credential could not access the repository.",
retryable=False,
)
if response.status_code >= 500:
raise ProviderError("UPSTREAM_ERROR", "Gitea returned a server error.", retryable=True)
if response.status_code != 200:
raise ProviderError(
"UPSTREAM_ERROR", "Gitea returned an unexpected response.", retryable=False,
)
try:
data = response.json()
except ValueError as exc:
raise ProviderError(
"UPSTREAM_ERROR",
"Gitea returned a response that could not be parsed.",
retryable=False,
) from exc
if not isinstance(data, dict):
raise ProviderError(
"UPSTREAM_ERROR", "Gitea returned an unexpected response shape.", retryable=False,
)
return data