diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index f5b9e9e..d6813a1 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -5291,6 +5291,7 @@ _RUNTIME_CAPABILITY_TASKS = ( "create_issue", "comment_issue", "create_pr", + "work_issue", "review_pr", "merge_pr", "close_issue", @@ -5343,6 +5344,7 @@ def _build_runtime_task_capabilities( "create_issue": "can_create_issues", "comment_issue": "can_comment_on_issues", "create_pr": "can_author_prs", + "work_issue": "can_work_issues", "review_pr": "can_review_prs", "merge_pr": "can_merge_prs", "close_issue": "can_close_issues", diff --git a/review_proofs.py b/review_proofs.py index 077883e..814fff0 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -2328,6 +2328,30 @@ CREATE_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS = ( ("Terminal review mutation", ("terminal review mutation",)), ) +WORK_ISSUE_WORKFLOW_MARKERS = ( + "workflows/work-issue.md", + "work-issue.md", +) + +WORK_ISSUE_TASK_MARKERS = ( + "work-issue", + "work issue", +) + +WORK_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS = ( + ("Selected PR", ("selected pr",)), + ("Review decision", ("review decision",)), + ("Merge result", ("merge result",)), + ("Merge preflight", ("merge preflight",)), + ("Already-landed gate", ("already-landed gate",)), + ("Pinned reviewed head", ("pinned reviewed head",)), + ("Terminal review mutation", ("terminal review mutation",)), + ("Duplicate search terms", ("duplicate search terms",)), + ("Issues created", ("issues created",)), + ("Issues skipped as duplicates", ("issues skipped as duplicates",)), + ("Requested issue task", ("requested issue task",)), +) + def _handoff_section_lines(report_text): """Return the lines of the Controller Handoff section, or None.""" @@ -3520,6 +3544,104 @@ def assess_create_issue_workflow_source(report_text: str) -> dict: } +def assess_work_issue_workflow_source(report_text: str) -> dict: + """#139: work-issue reports must cite the canonical workflow file.""" + lower = (report_text or "").lower() + reasons = [] + if not any(marker in lower for marker in WORK_ISSUE_WORKFLOW_MARKERS): + reasons.append( + "work-issue report missing workflow source " + "(workflows/work-issue.md)" + ) + if not any(marker in lower for marker in WORK_ISSUE_TASK_MARKERS): + reasons.append( + "work-issue report missing task mode declaration (work-issue)" + ) + return { + "complete": not reasons, + "downgraded": bool(reasons), + "reasons": reasons, + } + + +def assess_work_issue_mode_isolation(report_text: str) -> dict: + """#139: reject review-merge/create-issue handoff fields in work-issue runs.""" + section = _handoff_section_lines(report_text) + reasons = [] + if section is None: + return { + "complete": True, + "downgraded": False, + "reasons": [], + } + + labels = [] + for line in section: + stripped = line.strip().lstrip("-*").strip() + if ":" in stripped: + labels.append(stripped.split(":", 1)[0].strip().lower()) + + for name, aliases in WORK_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS: + if any( + label.startswith(alias) + for label in labels + for alias in aliases + ): + reasons.append( + f"work-issue handoff must not include cross-mode field " + f"'{name}'" + ) + + legacy_workspace = any( + label.startswith("workspace mutations") for label in labels + ) + precise_categories = any( + label.startswith("file edits by author") + for label in labels + ) + if legacy_workspace and not precise_categories: + reasons.append( + "work-issue handoff must use precise mutation categories " + "instead of legacy 'Workspace mutations'" + ) + + return { + "complete": not reasons, + "downgraded": bool(reasons), + "reasons": reasons, + } + + +def assess_work_issue_final_report(report_text: str) -> dict: + """#139: composite verifier for work-issue final reports.""" + checks = { + "workflow_source": assess_work_issue_workflow_source(report_text), + "mode_isolation": assess_work_issue_mode_isolation(report_text), + } + + reasons = [] + downgraded = False + for name, result in checks.items(): + verdict = result.get("verdict") + if verdict in ("missing", "incomplete"): + downgraded = True + reasons.extend(result.get("reasons") or []) + elif result.get("downgraded") or not result.get("complete", True): + downgraded = True + reasons.extend( + f"{name}: {r}" for r in (result.get("reasons") or []) + ) + + grade = "A" if not downgraded else "downgraded" + return { + "grade": grade, + "downgraded": downgraded, + "checks": checks, + "reasons": reasons, + "complete": not downgraded, + } + + def assess_create_issue_mode_isolation(report_text: str) -> dict: """#337: reject work-issue/review-merge handoff fields in create-issue runs.""" section = _handoff_section_lines(report_text) diff --git a/skills/llm-project-workflow/workflows/work-issue.md b/skills/llm-project-workflow/workflows/work-issue.md index 303b4cb..f450cd7 100644 --- a/skills/llm-project-workflow/workflows/work-issue.md +++ b/skills/llm-project-workflow/workflows/work-issue.md @@ -83,6 +83,20 @@ Examples: If capability cannot be proven, stop and produce a recovery handoff only. +### 2A. Pre-task role routing (#139) + +Before issue selection or any mutation, resolve the composite author task: + +* `gitea_route_task_session(task_type="work-issue", remote=…)` — must return + `route_result: allowed_current_session` and `downstream_allowed: true` +* `gitea_resolve_task_capability(task="work_issue", remote=…)` — must show + `required_role_kind: author` and `allowed_in_current_session: true` + +Hyphen (`work-issue`) and underscore (`work_issue`) aliases are equivalent. +If routing returns `ambiguous_task_stop`, `wrong_role_stop`, or +`route_to_author_session`, stop and produce a recovery handoff only — do not +fall back to reviewer tools or guess a different profile. + ## 3. Stop immediately on blocked infrastructure If any of the following appears, stop immediately: diff --git a/task_capability_map.py b/task_capability_map.py index fe00f26..1d40a78 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -104,6 +104,14 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.read", "role": "author", }, + "work_issue": { + "permission": "gitea.pr.create", + "role": "author", + }, + "work-issue": { + "permission": "gitea.pr.create", + "role": "author", + }, "reconcile_landed_pr": { "permission": "gitea.read", "role": "author", diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 4e56a15..f986cd8 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -100,6 +100,8 @@ def test_work_issue_workflow_contract(): assert "canonical: true" in text assert "Do not merge your own PR" in text assert "## 21. PR creation rules" in text + assert "gitea_route_task_session" in text + assert "work-issue" in text def test_pr_queue_cleanup_workflow_contract(): @@ -181,6 +183,12 @@ def test_create_issue_final_report_verifier_exported(): assert callable(assess_create_issue_final_report) +def test_work_issue_final_report_verifier_exported(): + from review_proofs import assess_work_issue_final_report + + assert callable(assess_work_issue_final_report) + + def test_merge_simulation_verifier_exported(): from review_proofs import assess_merge_simulation_report diff --git a/tests/test_resolve_task_capability.py b/tests/test_resolve_task_capability.py index d3a3b80..a34456d 100644 --- a/tests/test_resolve_task_capability.py +++ b/tests/test_resolve_task_capability.py @@ -121,6 +121,32 @@ class TestResolveTaskCapability(unittest.TestCase): self.assertTrue(res["allowed_in_current_session"]) self.assertFalse(res["different_mcp_namespace_required"]) + @patch("mcp_server.api_request", return_value={"login": "author-user"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_resolve_work_issue_author_profile_allowed(self, _auth, _api): + with patch.dict(os.environ, self._env("author-profile")): + for task in ("work_issue", "work-issue"): + res = mcp_server.gitea_resolve_task_capability( + task=task, remote="prgs" + ) + self.assertEqual(res["requested_task"], task) + self.assertEqual( + res["required_operation_permission"], "gitea.pr.create" + ) + self.assertEqual(res["required_role_kind"], "author") + self.assertTrue(res["allowed_in_current_session"]) + + @patch("mcp_server.api_request", return_value={"login": "author-user"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_resolve_work_issue_reviewer_profile_blocked(self, _auth, _api): + with patch.dict(os.environ, self._env("reviewer-profile")): + res = mcp_server.gitea_resolve_task_capability( + task="work_issue", remote="prgs" + ) + self.assertFalse(res["allowed_in_current_session"]) + self.assertEqual(res["required_role_kind"], "author") + self.assertTrue(res["different_mcp_namespace_required"]) + @patch("mcp_server.api_request", return_value={"login": "author-user"}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_resolve_issue_comment_task(self, _auth, _api): diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index 0558517..f132fc2 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -40,6 +40,9 @@ from review_proofs import ( # noqa: E402 assess_create_issue_final_report, assess_create_issue_mode_isolation, assess_create_issue_workflow_source, + assess_work_issue_final_report, + assess_work_issue_mode_isolation, + assess_work_issue_workflow_source, assess_edited_pr_inventory_coverage, assess_empty_queue_report, assess_fresh_issue_selection, @@ -2320,6 +2323,52 @@ class TestCreateIssueFinalReport(unittest.TestCase): self.assertTrue(result["downgraded"]) +class TestWorkIssueFinalReport(unittest.TestCase): + """#139: work-issue final-report verifier coverage.""" + + def _good_report(self): + return "\n".join([ + "Task mode: work-issue", + "Workflow source: skills/llm-project-workflow/workflows/work-issue.md", + "## Controller Handoff", + "- Task: work-issue", + "- Repo: Scaled-Tech-Consulting/Gitea-Tools", + "- Role: author", + "- Identity: jcwalker3 / prgs-author", + "- Active profile: prgs-author", + "- Runtime context: prgs-author on prgs", + "- Selected issue: #139", + "- Branch name: feat/issue-139-role-aware-work-issue-routing", + "- PR number: not created in this session", + "- File edits by author: role_session_router.py", + "- Blockers: none", + "- Current status: implementing routing", + "- Safe next action: open PR", + "- Next: open PR", + "- Safety statement: no review/merge", + ]) + + def test_complete_work_issue_report_earns_a(self): + result = assess_work_issue_final_report(self._good_report()) + self.assertEqual(result["grade"], "A") + self.assertFalse(result["downgraded"]) + + def test_missing_workflow_source_downgrades(self): + report = self._good_report().replace( + "workflows/work-issue.md", "SKILL.md only" + ) + result = assess_work_issue_workflow_source(report) + self.assertTrue(result["downgraded"]) + + def test_review_field_in_handoff_downgrades(self): + report = self._good_report().replace( + "- Safety statement:", + "- Review decision: APPROVE\n- Safety statement:", + ) + result = assess_work_issue_mode_isolation(report) + self.assertTrue(result["downgraded"]) + + class TestValidationEnvironmentProof(unittest.TestCase): """Issue #311: exact validation command and execution environment.""" diff --git a/tests/test_role_session_router.py b/tests/test_role_session_router.py index 35dc660..f545963 100644 --- a/tests/test_role_session_router.py +++ b/tests/test_role_session_router.py @@ -192,6 +192,31 @@ class TestRoleSessionRouter(unittest.TestCase): ) self.assertEqual(result.get("number"), 999) + @patch("mcp_server.api_request", return_value={"login": "jcwalker3"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_work_issue_task_under_author_profile_allowed(self, _auth, _api): + with patch.dict(os.environ, self._env("prgs-author")): + for task_type in ("work_issue", "work-issue"): + route = mcp_server.gitea_route_task_session( + task_type=task_type, remote="prgs" + ) + self.assertEqual( + route["route_result"], role_session_router.ROUTE_ALLOWED + ) + self.assertTrue(route["downstream_allowed"]) + + @patch("mcp_server.api_request", return_value={"login": "sysadmin"}) + @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") + def test_work_issue_task_under_reviewer_profile_routes_to_author(self, _auth, _api): + with patch.dict(os.environ, self._env("prgs-reviewer")): + route = mcp_server.gitea_route_task_session( + task_type="work-issue", remote="prgs" + ) + self.assertEqual( + route["route_result"], role_session_router.ROUTE_TO_AUTHOR + ) + self.assertFalse(route["downstream_allowed"]) + @patch("mcp_server.api_request", return_value={"login": "sysadmin"}) @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") def test_reviewer_task_under_reviewer_profile_allowed(self, _auth, _api):