fix: isolate review-mutation ledger from host session-state (#590 )

Tests that call patch.dict(os.environ, clear=True) dropped GITEA_MCP_SESSION_STATE_DIR and re-adopted the operator host cache under ~/.cache/gitea-tools/session-state. A residual terminal request_changes mutation then tripped the #332 hard-stop before test_unknown_profile_blocks could assert the empty-profile gate. - Pin mcp_session_state.default_state_dir / DEFAULT_STATE_DIR to a per-test temp dir in conftest (still honors an explicit env override) - Clear the decision lock at the start of each TestMergePR test - Add regression coverage for env-clear isolation Closes #590
2026-07-09 14:16:39 -04:00
4 changed files with 0 additions and 433 deletions
@@ -799,7 +799,6 @@ 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
@@ -1673,7 +1672,6 @@ 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.
@@ -1693,8 +1691,6 @@ 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:
@@ -1723,25 +1719,6 @@ 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,
@@ -1,180 +0,0 @@
 """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
@@ -344,41 +344,6 @@ 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:
@@ -1,195 +0,0 @@
 """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()