Merge pull request 'feat: block vague-reference issue creation with content preflight (Closes #582)' (#586) from feat/issue-582-issue-content-preflight into master

gitea_mcp_server.py

@@ -799,6 +799,7 @@ import gitea_audit  # noqa: E402
 import gitea_config  # noqa: E402
 import capability_stop_terminal  # noqa: E402
 import issue_duplicate_gate  # noqa: E402
+import issue_content_gate  # noqa: E402
 import role_session_router  # noqa: E402
 import role_namespace_gate  # noqa: E402
 import task_capability_map  # noqa: E402
@@ -1672,6 +1673,7 @@ def gitea_create_issue(
     require_workflow_labels: bool = False,
     allow_duplicate_override: bool = False,
     split_from_issue: int | None = None,
+    allow_incomplete_content: bool = False,
     worktree_path: str | None = None,
 ) -> dict:
     """Create a new issue on a Gitea repository.
@@ -1691,6 +1693,8 @@ def gitea_create_issue(
             type:* and one status:* label.
         allow_duplicate_override: Operator-approved split after duplicate found.
         split_from_issue: Existing duplicate issue number when overriding.
+        allow_incomplete_content: Operator override to bypass the #582 content
+            gate (create despite vague/missing title/body). Off by default.
         worktree_path: Optional path to verify branches-only guard.
 
     Returns:
@@ -1719,6 +1723,25 @@ def gitea_create_issue(
     if blocked:
         return blocked
     verify_preflight_purity(remote, worktree_path=worktree_path, task="create_issue")
+    content_gate = issue_content_gate.pre_create_issue_content_gate(
+        title,
+        body,
+        allow_incomplete=allow_incomplete_content,
+    )
+    if not content_gate.get("performed"):
+        return {
+            "success": False,
+            "performed": False,
+            "number": None,
+            "content_gate": content_gate,
+            "reasons": [
+                "BLOCKED + DIAGNOSE: issue content is a vague reference or "
+                "missing required fields; provide full title/body or a durable "
+                "source pointer (existing issue/PR/comment, checked-in file, "
+                "URL, or scratchpad path). "
+                + content_gate.get("diagnose", ""),
+            ],
+        }
     base = repo_api_url(h, o, r)
     requested_labels = issue_workflow_labels.labels_for_new_issue(
         issue_type=issue_type,

issue_content_gate.py

@@ -0,0 +1,180 @@
+"""Pre-create issue content gate (#582).
+
+Blocks issue creation when the title/body is a *vague reference* to
+out-of-band content the LLM cannot prove it has ("the drafted issue",
+"the prepared issue", "the issue we discussed", "the previous draft")
+with no durable source pointer.
+
+The rule from #582: an LLM must not fabricate an issue from stale chat
+memory. When the requested issue content is a vague reference and no
+durable source pointer (existing issue/PR/comment, checked-in file, URL,
+or scratchpad path) is present, the gate fails closed and the caller must
+return BLOCKED + DIAGNOSE naming the exact vague/missing fields.
+
+This gate intentionally does NOT require a non-empty body: title-only
+issue creation remains allowed for callers that explicitly want it. It
+only blocks content that is a placeholder reference to something the
+current context does not contain.
+"""
+
+from __future__ import annotations
+
+import re
+import unicodedata
+
+# Vague reference phrases that point at out-of-band draft content. Stored
+# normalized (lowercase, punctuation-stripped, whitespace-collapsed) so they
+# can be matched against normalized field text.
+VAGUE_REFERENCE_PHRASES: tuple[str, ...] = (
+    "the drafted issue",
+    "drafted issue",
+    "the prepared issue",
+    "prepared issue",
+    "the issue we discussed",
+    "the issue we talked about",
+    "the one we discussed",
+    "the previous draft",
+    "previous draft",
+    "the draft above",
+    "the issue above",
+    "as discussed",
+    "as we discussed",
+    "as previously discussed",
+    "per our discussion",
+    "per our conversation",
+    "per our chat",
+    "see above",
+    "same as before",
+    "the issue from earlier",
+    "the issue i drafted",
+    "the issue you drafted",
+)
+
+# A field that only restates a vague phrase (optionally with a leading verb
+# such as "create"/"add"/"open"/"file") is still a vague reference.
+_LEADING_VERBS = ("create", "add", "open", "file", "make", "please", "the")
+
+# Durable source pointers: if any of these appears, the content points at a
+# retrievable source of truth and is NOT treated as a fabricated reference.
+_DURABLE_SOURCE_REGEXES: tuple[re.Pattern[str], ...] = (
+    re.compile(r"#\d+"),                                   # #402
+    re.compile(r"\b(?:issue|pull|pr)\s+#?\d+", re.I),      # issue 402 / PR #17
+    re.compile(r"\bcomment\s+#?\d+", re.I),                # comment 8155
+    re.compile(r"https?://\S+", re.I),                     # URL
+    re.compile(                                            # checked-in file
+        r"[\w./-]+\.(?:py|md|json|txt|sh|ya?ml|toml|cfg|ini|rst)\b", re.I
+    ),
+    re.compile(r"\bscratchpad\b", re.I),                   # scratchpad path
+    re.compile(r"(?:^|\s)/[\w./-]+"),                      # absolute path
+)
+
+
+def normalize_text(text: str) -> str:
+    """Lowercase, punctuation-stripped, whitespace-collapsed text."""
+    norm = unicodedata.normalize("NFKC", (text or "").strip().lower())
+    norm = re.sub(r"[^\w\s]", " ", norm)
+    norm = re.sub(r"\s+", " ", norm).strip()
+    return norm
+
+
+def has_durable_source_pointer(text: str) -> bool:
+    """True when the raw text references a durable, retrievable source."""
+    raw = text or ""
+    return any(rx.search(raw) for rx in _DURABLE_SOURCE_REGEXES)
+
+
+def find_vague_reference(text: str) -> str | None:
+    """Return the vague phrase a field is dominated by, else ``None``.
+
+    A field is a vague reference only when it contains a known vague phrase,
+    carries no durable source pointer, and the phrase dominates the field
+    (the field is essentially just the phrase). Long, specific bodies that
+    merely contain "as discussed" in passing are not blocked.
+    """
+    if has_durable_source_pointer(text):
+        return None
+    norm = normalize_text(text)
+    if not norm:
+        return None
+    stripped = norm
+    for verb in _LEADING_VERBS:
+        if stripped.startswith(verb + " "):
+            stripped = stripped[len(verb) + 1:]
+    for phrase in VAGUE_REFERENCE_PHRASES:
+        if phrase in norm and len(stripped) <= len(phrase) + 12:
+            return phrase
+    return None
+
+
+def assess_issue_content(
+    title: str,
+    body: str = "",
+    *,
+    allow_incomplete: bool = False,
+) -> dict:
+    """Evaluate whether proposed issue content is durable enough to create.
+
+    Returns a dict with ``complete`` (bool), ``missing_fields``,
+    ``vague_fields``, ``reasons``, ``diagnose`` (joined reasons), and
+    ``has_durable_source_pointer``. ``allow_incomplete`` is an explicit
+    operator override that forces ``complete`` True.
+    """
+    t = (title or "").strip()
+    b = (body or "").strip()
+    missing_fields: list[str] = []
+    vague_fields: list[str] = []
+    reasons: list[str] = []
+
+    if not t:
+        missing_fields.append("title")
+        reasons.append("issue title is required")
+    else:
+        phrase = find_vague_reference(t)
+        if phrase:
+            vague_fields.append("title")
+            reasons.append(
+                f"title is a vague reference ('{phrase}') with no durable "
+                "content or source pointer"
+            )
+
+    if b:
+        phrase = find_vague_reference(b)
+        if phrase:
+            vague_fields.append("body")
+            reasons.append(
+                f"body is a vague reference ('{phrase}') with no durable "
+                "content or source pointer"
+            )
+
+    complete = allow_incomplete or (not missing_fields and not vague_fields)
+    return {
+        "complete": complete,
+        "missing_fields": missing_fields,
+        "vague_fields": vague_fields,
+        "reasons": reasons,
+        "diagnose": "; ".join(reasons),
+        "has_durable_source_pointer": has_durable_source_pointer(b)
+        or has_durable_source_pointer(t),
+        "override_applied": bool(
+            allow_incomplete and (missing_fields or vague_fields)
+        ),
+    }
+
+
+def pre_create_issue_content_gate(
+    title: str,
+    body: str = "",
+    *,
+    allow_incomplete: bool = False,
+) -> dict:
+    """Content gate wrapper mirroring the duplicate gate shape.
+
+    ``performed`` is True when the content is durable enough to create (or an
+    explicit override was supplied). When False, the caller must return
+    BLOCKED + DIAGNOSE and must not create the issue.
+    """
+    assessment = assess_issue_content(
+        title, body, allow_incomplete=allow_incomplete
+    )
+    assessment["performed"] = assessment["complete"]
+    return assessment

skills/llm-project-workflow/workflows/create-issue.md

@@ -344,6 +344,41 @@ If edits are needed, make the smallest correction necessary and report the corre
 
 Do not silently change requested meaning.
 
+## 14a. Enforced content preflight (#582)
+
+`gitea_create_issue` runs a **content preflight gate** before duplicate
+search or creation. It fails closed (returns `BLOCKED + DIAGNOSE`, no issue
+created) when the title/body is a *vague reference* to out-of-band draft
+content the session cannot prove it holds, and no durable source pointer is
+present.
+
+An LLM must never fabricate issue content from stale chat memory. If asked to
+create "the drafted issue" / "the prepared issue" / "the issue we discussed"
+/ "the previous draft" without the full content in the current context or a
+durable source pointer, stop and return `BLOCKED + DIAGNOSE` naming the exact
+vague/missing fields.
+
+A **durable source pointer** is one of: an existing issue/PR reference
+(`#582`), a comment id (`comment 8155`), a checked-in file path
+(`task_capability_map.py`), a URL, or a scratchpad path. When a pointer is
+present, retrieve the real content from it before creating.
+
+The gate does **not** require a non-empty body; explicit title-only creation
+stays allowed. It only blocks placeholder references. An operator may bypass
+it with `allow_incomplete_content=True` (report the override).
+
+**Invalid prompts (must block):**
+
+* "Create the drafted issue."
+* "File the prepared issue we discussed."
+* "Open the previous draft."
+
+**Valid prompts (proceed):**
+
+* Full title + body + acceptance criteria supplied inline.
+* "Create the issue drafted in `#582`." (durable pointer → fetch and use it)
+* "Create the issue from `scratchpad/issue-draft.md`." (durable file pointer)
+
 ## 15. Labels, assignees, and metadata
 
 Apply labels, assignees, milestones, or project fields only if:

tests/test_issue_content_gate.py

@@ -0,0 +1,195 @@
+"""Tests for the pre-create issue content gate (Issue #582)."""
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+from issue_content_gate import (  # noqa: E402
+    assess_issue_content,
+    find_vague_reference,
+    has_durable_source_pointer,
+    normalize_text,
+    pre_create_issue_content_gate,
+)
+from mcp_server import gitea_create_issue  # noqa: E402
+
+FAKE_AUTH = "Basic dGVzdDp0ZXN0"
+ISSUE_WRITE_ENV = {
+    "GITEA_ALLOWED_OPERATIONS": "gitea.issue.create",
+}
+
+
+class TestNormalizeAndPointers(unittest.TestCase):
+
+    def test_normalize_strips_punctuation_and_case(self):
+        self.assertEqual(normalize_text("  The Drafted-Issue!  "), "the drafted issue")
+
+    def test_issue_ref_is_durable_pointer(self):
+        self.assertTrue(has_durable_source_pointer("See #402 for the spec"))
+
+    def test_file_path_is_durable_pointer(self):
+        self.assertTrue(has_durable_source_pointer("Draft in task_capability_map.py"))
+
+    def test_scratchpad_is_durable_pointer(self):
+        self.assertTrue(has_durable_source_pointer("Full text in scratchpad/draft.md"))
+
+    def test_url_is_durable_pointer(self):
+        self.assertTrue(
+            has_durable_source_pointer("https://gitea.prgs.cc/issues/1 has it")
+        )
+
+    def test_plain_text_has_no_durable_pointer(self):
+        self.assertFalse(has_durable_source_pointer("just some words here"))
+
+
+class TestFindVagueReference(unittest.TestCase):
+
+    def test_bare_vague_phrases_are_flagged(self):
+        for text in (
+            "the drafted issue",
+            "The prepared issue",
+            "the issue we discussed",
+            "the previous draft",
+            "Create the drafted issue",
+            "please file the prepared issue",
+        ):
+            with self.subTest(text=text):
+                self.assertIsNotNone(find_vague_reference(text))
+
+    def test_durable_pointer_defeats_vague_reference(self):
+        # Same vague phrase but with a durable pointer -> not vague.
+        self.assertIsNone(find_vague_reference("the drafted issue in #402"))
+        self.assertIsNone(
+            find_vague_reference("the prepared issue: see scratchpad/draft.md")
+        )
+
+    def test_long_specific_body_with_incidental_phrase_is_not_vague(self):
+        body = (
+            "As discussed, the create_issue tool must reject vague references. "
+            "Add a preflight check that lists the exact missing fields and "
+            "returns BLOCKED + DIAGNOSE with concrete acceptance criteria."
+        )
+        self.assertIsNone(find_vague_reference(body))
+
+    def test_concrete_short_body_is_not_vague(self):
+        self.assertIsNone(find_vague_reference("Fix null deref in auth handler"))
+
+    def test_empty_is_not_vague(self):
+        self.assertIsNone(find_vague_reference(""))
+
+
+class TestAssessIssueContent(unittest.TestCase):
+
+    def test_missing_title_is_incomplete(self):
+        result = assess_issue_content("", "some body")
+        self.assertFalse(result["complete"])
+        self.assertIn("title", result["missing_fields"])
+
+    def test_vague_body_is_incomplete_and_diagnosed(self):
+        result = assess_issue_content("Add the thing", "the drafted issue")
+        self.assertFalse(result["complete"])
+        self.assertIn("body", result["vague_fields"])
+        self.assertIn("vague reference", result["diagnose"])
+
+    def test_vague_title_is_incomplete(self):
+        result = assess_issue_content("the prepared issue", "")
+        self.assertFalse(result["complete"])
+        self.assertIn("title", result["vague_fields"])
+
+    def test_complete_content_passes(self):
+        result = assess_issue_content(
+            "Block vague issue creation",
+            "Add a preflight gate that rejects placeholder references.",
+        )
+        self.assertTrue(result["complete"])
+        self.assertEqual(result["missing_fields"], [])
+        self.assertEqual(result["vague_fields"], [])
+
+    def test_title_only_is_allowed(self):
+        # Empty body must NOT block: title-only creation stays allowed.
+        result = assess_issue_content("A concrete specific title", "")
+        self.assertTrue(result["complete"])
+
+    def test_durable_pointer_body_passes(self):
+        result = assess_issue_content(
+            "Implement the drafted change",
+            "the drafted issue is fully specified in #582",
+        )
+        self.assertTrue(result["complete"])
+
+    def test_allow_incomplete_override(self):
+        result = assess_issue_content(
+            "the drafted issue", "the drafted issue", allow_incomplete=True
+        )
+        self.assertTrue(result["complete"])
+        self.assertTrue(result["override_applied"])
+
+    def test_gate_wrapper_sets_performed(self):
+        blocked = pre_create_issue_content_gate("Add x", "the drafted issue")
+        self.assertFalse(blocked["performed"])
+        ok = pre_create_issue_content_gate("Add x", "concrete actionable body")
+        self.assertTrue(ok["performed"])
+
+
+class TestCreateIssueContentMCPGate(unittest.TestCase):
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop")
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_blocks_vague_reference_without_creating(
+        self, _auth, mock_get_all, mock_api, mock_role_check
+    ):
+        mock_role_check.return_value = (True, [])
+        with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
+            result = gitea_create_issue(
+                title="Create the drafted issue",
+                body="the drafted issue",
+                remote="prgs",
+            )
+        self.assertFalse(result["performed"])
+        self.assertIn("content_gate", result)
+        self.assertIn("body", result["content_gate"]["vague_fields"])
+        self.assertTrue(any("BLOCKED + DIAGNOSE" in r for r in result["reasons"]))
+        mock_api.assert_not_called()
+        mock_get_all.assert_not_called()
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop")
+    @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_creates_when_content_is_concrete(
+        self, _auth, _get_all, mock_api, mock_role_check
+    ):
+        mock_role_check.return_value = (True, [])
+        mock_api.return_value = {"number": 7, "html_url": "https://x/issues/7"}
+        with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
+            result = gitea_create_issue(
+                title="Block vague issue creation",
+                body="Add a preflight gate rejecting placeholder references.",
+                remote="prgs",
+            )
+        self.assertEqual(result["number"], 7)
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop")
+    @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_override_creates_despite_vague_content(
+        self, _auth, _get_all, mock_api, mock_role_check
+    ):
+        mock_role_check.return_value = (True, [])
+        mock_api.return_value = {"number": 8, "html_url": "https://x/issues/8"}
+        with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
+            result = gitea_create_issue(
+                title="the drafted issue",
+                body="the drafted issue",
+                remote="prgs",
+                allow_incomplete_content=True,
+            )
+        self.assertEqual(result["number"], 8)
+
+
+if __name__ == "__main__":
+    unittest.main()