Compare commits

...
5 Commits
Author SHA1 Message Date
sysadminandClaude Fable 5 45c5cac2bc Require and validate Controller Handoff sections in workflow final reports
- Upgrade SKILL.md §K compact format to the issue #182 canonical field set
  (Task/Repo/Role/Identity/Issue-PR/Branch-SHA/Files/Validation/Mutations/
  Current status/Blockers/Next/Safety) plus role-specific field lists for
  review/merge, author, and queue/inventory tasks.
- Point the review-pr, merge-pr, and start-issue template handoff lines at
  the exactly-titled Controller Handoff section with their role fields.
- Add review_proofs.assess_controller_handoff(): reports without the exact
  section are 'missing' (downgraded), present-but-partial are 'incomplete'
  with the absent fields listed, and role extras are enforced per role.
- Add TestControllerHandoff (8 tests) including a SKILL.md doc-contract
  test so the documented requirement cannot silently rot.

The handoff supplements the full report; full-report validation rules are
unchanged.

Closes #182

Co-Authored-By: Claude Fable 5 <[email protected]>
2026-07-05 15:36:57 -04:00
sysadmin 601c608c58 Merge pull request 'Add repo-name disambiguation and complete-inventory proofs for blind PR review' (#185) from feat/issue-184-repo-name-disambiguation into master 2026-07-05 14:32:34 -05:00
sysadmin e2bccbafee feat(review-workflow): add repo-name disambiguation and complete-inventory proofs (#184) 2026-07-05 15:30:06 -04:00
sysadmin 340b585bfd Merge pull request 'test: isolate CLI output capture with monkeypatch/redirect to fix stdout corruption in full suite (Issue #178)' (#180) from feat/issue-178-fix-test-suite-stdout-capture into master 2026-07-05 14:26:31 -05:00
sysadmin ff4ab500df test: isolate CLI output capture with monkeypatch/redirect to fix stdout corruption in full suite runs (Issue #178)
- Replace raw patch("sys.stdout") and contextlib.redirect_stderr with pytest MonkeyPatch or contextlib redirect scoped to main calls in CLI tests (create_*, prs, python_cli, manage_labels, merge_pr, review_pr).
- Add regression tests in test_review_proofs.py for stdout/stderr remaining usable.
- Ensures normal pytest summary output without junitxml workaround; improves review validation reporting per #173.
- No behavior change to production code or gates.

Refs #178
2026-07-05 15:16:55 -04:00
14 changed files with 536 additions and 51 deletions
+172
View File
@@ -18,6 +18,54 @@ import re
_FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
# Repo name disambiguation rules for blind PR queue review.
# User phrases referencing the "MCP Gitea tool" (or similar) must resolve to
# Gitea-Tools repo, not be confused with mcp-control-plane.
# If ambiguous (e.g. just "open PRs"), check both configured repos.
REPO_ALIASES = {
"gitea-tools": "Scaled-Tech-Consulting/Gitea-Tools",
"gitea tool": "Scaled-Tech-Consulting/Gitea-Tools",
"mcp gitea tool": "Scaled-Tech-Consulting/Gitea-Tools",
"gitea mcp tool": "Scaled-Tech-Consulting/Gitea-Tools",
"gitea-tools repo": "Scaled-Tech-Consulting/Gitea-Tools",
"mcp-control-plane": "Scaled-Tech-Consulting/mcp-control-plane",
"mcp control plane": "Scaled-Tech-Consulting/mcp-control-plane",
}
def resolve_repos_from_user_reference(
reference: str, configured: list[str] | None = None
) -> list[str]:
"""Resolve a user reference string to the list of target repos to inventory.
- Exact aliases for Gitea-Tools map only to Gitea-Tools.
- mcp-control-plane aliases map only to it.
- Empty, ambiguous, or general "open PRs" default to all configured repos
(both by default).
- Returns subset of configured; never invents new repos.
"""
if configured is None:
configured = [
"Scaled-Tech-Consulting/Gitea-Tools",
"Scaled-Tech-Consulting/mcp-control-plane",
]
if not reference or not reference.strip():
return list(configured)
ref_lower = reference.lower()
matched = []
for alias, full_name in REPO_ALIASES.items():
if alias in ref_lower:
if full_name not in matched:
matched.append(full_name)
if matched:
# return only the matched ones that are in configured, preserving order
return [r for r in configured if r in matched]
# no specific alias match → check all (complete inventory required)
return list(configured)
SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION = (
"evidence missing: report contamination as unknown and "
"choose another PR or stop"
@@ -361,3 +409,127 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"merge_performed": bool(merge_performed),
"issue_status_verified": bool(issue_status_verified),
}
# ── Controller Handoff validation (Issue #182) ────────────────────────────────
#
# Every final report must end with a compact section titled exactly
# "Controller Handoff". Each required field is a (canonical name, aliases)
# pair; a field counts as present when any alias starts a bullet/label line
# inside the handoff section.
HANDOFF_HEADING = "Controller Handoff"
HANDOFF_BASE_FIELDS = (
("Task", ("task",)),
("Repo", ("repo", "repository", "repo/state")),
("Role", ("role",)),
("Identity", ("identity",)),
("Issue/PR", ("issue/pr", "issues/prs", "issue", "pr")),
("Branch/SHA", ("branch/sha", "branch", "head sha")),
("Files changed", ("files changed", "changed", "files")),
("Validation", ("validation",)),
("Mutations", ("mutations",)),
("Current status", ("current status", "status")),
("Blockers", ("blockers",)),
("Next", ("next",)),
("Safety", ("safety",)),
)
HANDOFF_ROLE_FIELDS = {
"review": (
("Selected PR", ("selected pr",)),
("Reviewer eligibility", ("reviewer eligibility", "eligibility")),
("Pinned reviewed head", ("pinned reviewed head", "pinned head")),
("Review decision", ("review decision", "decision")),
("Merge result", ("merge result",)),
("Linked issue status", ("linked issue status", "linked issue")),
("Cleanup status", ("cleanup status", "cleanup")),
),
"author": (
("Selected issue", ("selected issue",)),
("Claim/comment status", ("claim/comment status", "claim status",
"claim")),
("PR number opened", ("pr number opened", "pr opened", "pr number")),
("No review/merge confirmation", ("no review/merge",
"no review or merge")),
),
"inventory": (
("Repositories checked", ("repositories checked", "repos checked")),
("Open PR counts", ("open pr counts", "open pr count",
"open prs per repo")),
("Selected PR or reason", ("selected pr", "none selected",
"reason none selected")),
("Inventory completeness", ("inventory complete", "inventory scoped",
"inventory completeness")),
),
}
def _handoff_section_lines(report_text):
"""Return the lines of the Controller Handoff section, or None."""
lines = (report_text or "").splitlines()
start = None
for i, line in enumerate(lines):
bare = line.strip().lstrip("#").strip().rstrip(":")
if bare == HANDOFF_HEADING:
start = i + 1
break
if start is None:
return None
return lines[start:]
def assess_controller_handoff(report_text, role=None):
"""Issue #182: final reports without a Controller Handoff downgrade.
Verdicts:
- 'missing' — no exactly-titled section; the report is downgraded.
- 'incomplete' — section present but required fields absent (listed).
- 'complete' — all base fields plus the role-specific fields present.
*role* is 'review', 'author', 'inventory', or None (base fields only).
The handoff supplements the full report; this helper never validates
the full report body, only the continuation summary.
"""
section = _handoff_section_lines(report_text)
if section is None:
return {
"verdict": "missing",
"downgraded": True,
"missing_fields": [name for name, _ in HANDOFF_BASE_FIELDS],
"reasons": [
"final report has no section titled exactly "
f"'{HANDOFF_HEADING}'"
],
}
labels = []
for line in section:
stripped = line.strip().lstrip("-*").strip()
if ":" in stripped:
labels.append(stripped.split(":", 1)[0].strip().lower())
required = list(HANDOFF_BASE_FIELDS)
required.extend(HANDOFF_ROLE_FIELDS.get(role or "", ()))
missing = []
for name, aliases in required:
if not any(label.startswith(alias)
for label in labels for alias in aliases):
missing.append(name)
if missing:
return {
"verdict": "incomplete",
"downgraded": True,
"missing_fields": missing,
"reasons": [f"handoff missing required field: {m}"
for m in missing],
}
return {
"verdict": "complete",
"downgraded": False,
"missing_fields": [],
"reasons": [],
}
+41 -10
View File
@@ -181,11 +181,21 @@ Worktree folder = branch with `/` replaced by `-`
that the diff base is the PR base branch. If `HEAD` does not match the
pinned head, **stop before review/merge**
(`review_proofs.verify_pinned_head_checkout`).
6. **Inventory proof (#173):** a blind queue review must prove listing
completeness before claiming "only PRs found": both configured
repositories checked, open-PR filters stated, pagination handled or
explicitly not needed, and the total open PR count per repo reported
(`review_proofs.assess_inventory_completeness`).
6. **Inventory proof (#173 + repo disambiguation hardening):** a blind queue
review must prove listing completeness before claiming "only PRs found".
Use repo-name disambiguation:
- "Gitea-Tools" / "gitea tool" / "MCP Gitea tool" / "gitea MCP tool" /
"gitea-tools repo" resolve **only** to `Scaled-Tech-Consulting/Gitea-Tools`.
- "mcp-control-plane" resolves only to `Scaled-Tech-Consulting/mcp-control-plane`.
- Ambiguous ("open PRs", no explicit repo, "MCP Gitea tooling") → inventory
**both** configured repos.
Report must state exactly which repo(s) were checked. If only one checked:
"Only <repo> was checked. Other configured repos were not checked. This is
not a complete queue inventory." Never let a single-repo zero hide PRs in
the other.
Both configured repos must be reported with state filter, pagination proof,
and open-PR count (`review_proofs.assess_inventory_completeness` and
`resolve_repos_from_user_reference`).
7. Inspect the full diff; confirm scope matches the linked issue; flag unrelated files.
8. Run the tests. Validation reporting must include the exact command and
exact results: pass/fail, counts of tests passed/skipped/failed, any
@@ -286,22 +296,43 @@ current state immediately, without rereading the conversation.
readability, not as a full human status report. PR bodies still carry the
full review detail — the handoff never replaces PR documentation.
Compact format (default):
Compact format (default, canonical field set per issue #182):
```md
## Controller Handoff
- Task:
- Repo/state:
- Issues/PRs:
- Changed:
- Repo:
- Role:
- Identity:
- Issue/PR:
- Branch/SHA:
- Files changed:
- Validation:
- Mutations:
- Current status:
- Blockers:
- Review:
- Next:
- Safety:
```
Role-specific fields (append to the compact block):
- review/merge tasks: `Selected PR:`, `Reviewer eligibility:`,
`Pinned reviewed head:`, `Review decision:`, `Merge result:`,
`Linked issue status:`, `Cleanup status:`
- author tasks: `Selected issue:`, `Claim/comment status:`,
`PR number opened:`, `No review/merge:` (explicit confirmation)
- queue/inventory tasks: `Repositories checked:`, `Open PR counts:`,
`Selected PR or reason none selected:`, `Inventory completeness:`
The section title must be exactly `Controller Handoff`.
`review_proofs.assess_controller_handoff()` validates this section; reports
missing it (or missing required fields) are downgraded. The handoff never
replaces the full report — it is the compact continuation summary at the end,
and the full report must still carry exact validation results and mutation
confirmation.
The `Safety:` line is never omitted; it is usually:
```text
@@ -35,5 +35,11 @@ Then run the cleanup template (worktree-cleanup.md):
- delete remote branch, remove local branch + worktree folder
- fetch/prune; confirm main checkout is clean and current (0 0).
Handoff: reviewer identity, merge result + commit, cleanup done, issue closed, PR metadata state/merged flag/hash, remote master hash, post-merge verification method used & verification results.
Handoff: end with a section titled exactly `Controller Handoff` per SKILL.md
§K (long form — a merge is always high-risk), including the review/merge role
fields (Selected PR, Reviewer eligibility, Pinned reviewed head, Review
decision, Merge result, Linked issue status, Cleanup status) plus: merge
commit, PR metadata state/merged flag/hash, remote master hash, and the
post-merge verification method used & verification results. Reports missing
the handoff are downgraded (review_proofs.assess_controller_handoff).
```
@@ -5,6 +5,19 @@ Copy, fill the `<...>` fields, and paste as the task prompt.
```text
Task: review PR #<pr> for issue #<n>.
Repo name disambiguation (Gitea-Tools blind review hardening):
- "Gitea-Tools", "gitea tool", "MCP Gitea tool", "gitea MCP tool", "gitea-tools repo"
→ MUST resolve to `Scaled-Tech-Consulting/Gitea-Tools` (never treat as mcp-control-plane).
- "mcp-control-plane", "mcp control plane" → only `Scaled-Tech-Consulting/mcp-control-plane`.
- If user says "open PRs", "the queue", "MCP Gitea tooling" without explicit repo,
or reference is ambiguous: check BOTH configured repos:
`Scaled-Tech-Consulting/Gitea-Tools` and `Scaled-Tech-Consulting/mcp-control-plane`.
- In the final report, always state exactly which repo(s) were checked.
If only one was checked: explicitly say "Only <repo> was checked. Other
configured repos were not checked. This is not a complete queue inventory."
- A single-repo "no open PRs" result MUST NOT be reported as global "no open PRs"
if the other configured repo was not inventoried.
Rules (llm-project-workflow):
- Review in a SEPARATE detached review worktree, never the author's folder.
- You must NOT be the PR author. If the authenticated user == PR author, stop.
@@ -52,7 +65,10 @@ Steps:
- MCP-Profile: <profile name>
- Eligibility: passed/failed
Handoff: reviewer identity, PR author, scope verdict, checks + results, decision —
formatted per SKILL.md §K (compact by default; long form if a merge happened
or a gate blocked you); if you could not merge, name the exact gate.
Handoff: end with a section titled exactly `Controller Handoff` per SKILL.md
§K (compact by default; long form if a merge happened or a gate blocked you),
including the review/merge role fields: Selected PR, Reviewer eligibility,
Pinned reviewed head, Review decision, Merge result, Linked issue status,
Cleanup status. If you could not merge, name the exact gate. Reports missing
the handoff are downgraded (review_proofs.assess_controller_handoff).
```
@@ -42,7 +42,10 @@ Steps:
- Self-review allowed: no
9. Stop before review/merge — you are the author.
Handoff: issue #, branch, worktree path, files changed, checks + results, PR URL —
formatted as the compact Controller Handoff (SKILL.md §K; long form only on
the high-risk triggers); Review line: "Review needed — PR is open".
Handoff: end with a section titled exactly `Controller Handoff` per SKILL.md
§K (compact; long form only on the high-risk triggers), including the author
role fields: Selected issue, Claim/comment status, PR number opened, and an
explicit "No review/merge:" confirmation — plus branch, worktree path, files
changed, checks + results. Next line: "Review needed — PR is open". Reports
missing the handoff are downgraded (review_proofs.assess_controller_handoff).
```
+9 -4
View File
@@ -11,6 +11,7 @@ import json
import sys
import tempfile
import unittest
import contextlib
from unittest.mock import MagicMock, patch
# The module under test lives in the repo root, not a package.
@@ -38,7 +39,8 @@ class TestArgParsing(unittest.TestCase):
@patch("create_issue.api_request", return_value={"number": 1, "html_url": "http://x/1"})
@patch("create_issue.get_credentials", return_value=FAKE_CREDS)
def test_minimal_args(self, _cred, _api):
rc = create_issue.main(["--title", "Hello"])
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
rc = create_issue.main(["--title", "Hello"])
self.assertEqual(rc, 0)
def test_missing_title_exits(self):
@@ -50,7 +52,8 @@ class TestArgParsing(unittest.TestCase):
@patch("create_issue.get_credentials", return_value=FAKE_CREDS)
def test_remote_choices(self, _cred, _api):
for remote in ("dadeschools", "prgs"):
rc = create_issue.main(["--remote", remote, "--title", "X"])
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
rc = create_issue.main(["--remote", remote, "--title", "X"])
self.assertEqual(rc, 0, f"--remote {remote} should be accepted")
def test_invalid_remote_exits(self):
@@ -138,7 +141,8 @@ class TestAuthFailure(unittest.TestCase):
@patch("create_issue.get_credentials", return_value=("", ""))
def test_no_credentials_returns_1(self, _cred):
rc = create_issue.main(["--title", "T"])
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
rc = create_issue.main(["--title", "T"])
self.assertEqual(rc, 1)
@@ -152,7 +156,8 @@ class TestAPIError(unittest.TestCase):
def test_api_error_returns_1(self, _cred):
with patch("create_issue.api_request",
side_effect=RuntimeError("HTTP 422: duplicate")):
rc = create_issue.main(["--title", "Dup"])
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
rc = create_issue.main(["--title", "Dup"])
self.assertEqual(rc, 1)
+3 -1
View File
@@ -7,6 +7,7 @@ import io
import json
import sys
import unittest
import contextlib
from unittest.mock import MagicMock, patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
@@ -34,7 +35,8 @@ class TestArgParsing(unittest.TestCase):
@patch("create_pr.urllib.request.urlopen", return_value=_mock_urlopen())
@patch("create_pr.get_credentials", return_value=FAKE_CREDS)
def test_minimal_required_args(self, _cred, _url):
rc = create_pr.main(["--title", "PR Title", "--head", "feat/branch"])
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
rc = create_pr.main(["--title", "PR Title", "--head", "feat/branch"])
self.assertEqual(rc, 0)
def test_missing_title_exits(self):
+10 -4
View File
@@ -2,9 +2,11 @@
All API calls are mocked — no real network or keychain access.
"""
import io
import json
import sys
import unittest
import contextlib
from unittest.mock import MagicMock, call, patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
@@ -33,7 +35,8 @@ class TestLabelCreation(unittest.TestCase):
# Patch sys.argv to avoid --dry
with patch.object(sys, "argv", ["manage_labels.py"]):
manage_labels.main()
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
manage_labels.main()
# The GET call happens, but no POST calls for label creation
get_calls = [c for c in mock_api.call_args_list if c[0][0] == "GET"]
@@ -59,7 +62,8 @@ class TestLabelCreation(unittest.TestCase):
mock_api.side_effect = side_effect
with patch.object(sys, "argv", ["manage_labels.py"]):
manage_labels.main()
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
manage_labels.main()
post_calls = [
c for c in mock_api.call_args_list
@@ -79,7 +83,8 @@ class TestDryRun(unittest.TestCase):
mock_api.return_value = [] # no existing labels
with patch.object(sys, "argv", ["manage_labels.py", "--dry"]):
manage_labels.main()
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
manage_labels.main()
# Only the GET call should be made, no POST or PUT
for c in mock_api.call_args_list:
@@ -107,7 +112,8 @@ class TestLabelMapping(unittest.TestCase):
mock_api.side_effect = side_effect
with patch.object(sys, "argv", ["manage_labels.py"]):
manage_labels.main()
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
manage_labels.main()
put_calls = [c for c in mock_api.call_args_list if c[0][0] == "PUT"]
self.assertEqual(len(put_calls), len(manage_labels.MAPPING))
+6 -2
View File
@@ -45,13 +45,17 @@ class TestMergeDisabled(unittest.TestCase):
mock_api.assert_not_called()
def test_message_points_to_gated_workflow(self):
from _pytest.monkeypatch import MonkeyPatch
import io
import contextlib
with patch("merge_pr.get_auth_header", return_value=FAKE_CREDS), \
patch("merge_pr.api_request") as mock_api:
buf = io.StringIO()
with contextlib.redirect_stderr(buf):
monkeypatch = MonkeyPatch()
monkeypatch.setattr(sys, "stderr", buf)
try:
rc = merge_pr.main(["--pr-number", "81"])
finally:
monkeypatch.undo()
self.assertEqual(rc, 2)
mock_api.assert_not_called()
msg = buf.getvalue().lower()
+29 -17
View File
@@ -9,6 +9,8 @@ import shutil
from unittest.mock import patch
from io import StringIO
from _pytest.monkeypatch import MonkeyPatch
# Add project root to sys.path
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if PROJECT_ROOT not in sys.path:
@@ -127,24 +129,29 @@ class TestMigrateProfiles(unittest.TestCase):
v2_data = migrate_profiles.migrate_v1_to_v2(self.v1_content)
self.assertTrue(migrate_profiles.validate_v2_data(v2_data))
@patch("sys.stdout", new_callable=StringIO)
def test_dry_run_default(self, mock_stdout):
def test_dry_run_default(self):
"""Verify that running without -w prints generated config without modifying files."""
output_file = os.path.join(self.temp_dir, "migrated_dry.json")
test_args = [
"migrate_profiles.py",
"-i", self.input_file,
"-o", output_file
]
with patch.object(sys, "argv", test_args):
with self.assertRaises(SystemExit) as cm:
migrate_profiles.main()
self.assertEqual(cm.exception.code, 0)
monkeypatch = MonkeyPatch()
mock_stdout = StringIO()
monkeypatch.setattr(sys, "stdout", mock_stdout)
try:
output_file = os.path.join(self.temp_dir, "migrated_dry.json")
test_args = [
"migrate_profiles.py",
"-i", self.input_file,
"-o", output_file
]
with patch.object(sys, "argv", test_args):
with self.assertRaises(SystemExit) as cm:
migrate_profiles.main()
self.assertEqual(cm.exception.code, 0)
self.assertFalse(os.path.exists(output_file))
self.assertFalse(os.path.exists(f"{self.input_file}.bak"))
self.assertFalse(os.path.exists(output_file))
self.assertFalse(os.path.exists(f"{self.input_file}.bak"))
stdout_output = mock_stdout.getvalue()
stdout_output = mock_stdout.getvalue()
finally:
monkeypatch.undo()
self.assertIn("DRY-RUN MODE", stdout_output)
self.assertIn("version", stdout_output)
self.assertIn("identities", stdout_output)
@@ -165,13 +172,18 @@ class TestMigrateProfiles(unittest.TestCase):
json.dump(sensitive, f)
test_args = ["migrate_profiles.py", "-i", self.input_file]
with patch("sys.stdout", new_callable=StringIO) as mock_stdout:
monkeypatch = MonkeyPatch()
mock_stdout = StringIO()
monkeypatch.setattr(sys, "stdout", mock_stdout)
try:
with patch.object(sys, "argv", test_args):
with self.assertRaises(SystemExit) as cm:
migrate_profiles.main()
self.assertEqual(cm.exception.code, 0)
stdout_output = mock_stdout.getvalue()
stdout_output = mock_stdout.getvalue()
finally:
monkeypatch.undo()
self.assertNotIn("super-secret-token-value", stdout_output)
self.assertNotIn("token", stdout_output.lower())
+6 -2
View File
@@ -4,6 +4,8 @@ Mocks api_request and credentials.
"""
import sys
import unittest
import io
import contextlib
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
@@ -24,14 +26,16 @@ class TestListPRs(unittest.TestCase):
mock_api.return_value = [
{"number": 1, "title": "PR 1", "head": {"ref": "branch1"}, "base": {"ref": "main"}, "html_url": "http://url1", "mergeable": True}
]
rc = list_prs.main([])
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
rc = list_prs.main([])
self.assertEqual(rc, 0)
mock_api.assert_called_once()
@patch("list_prs.api_request", return_value=[])
@patch("list_prs.get_auth_header", return_value=FAKE_CREDS)
def test_list_prs_empty(self, _auth, mock_api):
rc = list_prs.main(["--state", "closed"])
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
rc = list_prs.main(["--state", "closed"])
self.assertEqual(rc, 0)
mock_api.assert_called_once()
+6 -2
View File
@@ -4,6 +4,8 @@ All tests mock credentials and API requests so no real network calls are made.
"""
import sys
import unittest
import io
import contextlib
from unittest.mock import patch, MagicMock
# The modules under test live in the repo root
@@ -31,7 +33,8 @@ class TestCloseIssueCLI(unittest.TestCase):
@patch("close_issue.api_request")
@patch("close_issue.get_auth_header", return_value=FAKE_AUTH)
def test_successful_close(self, _auth, mock_api):
rc = close_issue.main(["42"])
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
rc = close_issue.main(["42"])
self.assertEqual(rc, 0)
mock_api.assert_called_once()
url = mock_api.call_args[0][1]
@@ -69,7 +72,8 @@ class TestMarkIssueCLI(unittest.TestCase):
[{"id": 101, "name": "status:in-progress"}],
[{"name": "status:in-progress"}],
]
rc = mark_issue.main(["15", "start"])
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
rc = mark_issue.main(["15", "start"])
self.assertEqual(rc, 0)
self.assertEqual(mock_api.call_count, 2)
+6 -2
View File
@@ -79,15 +79,19 @@ class TestAPIPayload(unittest.TestCase):
self.assertEqual(mock_api.call_count, 0)
def test_merge_flag_message_points_to_gated_workflow(self):
from _pytest.monkeypatch import MonkeyPatch
import io
import contextlib
with patch("review_pr.get_auth_header", return_value=FAKE_CREDS), \
patch("review_pr.api_request") as mock_api:
buf = io.StringIO()
with contextlib.redirect_stderr(buf):
monkeypatch = MonkeyPatch()
monkeypatch.setattr(sys, "stderr", buf)
try:
rc = review_pr.main([
"--pr-number", "81", "--event", "APPROVE", "--merge",
])
finally:
monkeypatch.undo()
self.assertEqual(rc, 2)
self.assertEqual(mock_api.call_count, 0)
msg = buf.getvalue().lower()
+216
View File
@@ -14,16 +14,19 @@ each precondition of a blind queue review:
These are the harness assertions from the issue's Required behavior 7.
"""
import io
import sys
import unittest
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
from review_proofs import ( # noqa: E402
assess_controller_handoff,
assess_inventory_completeness,
assess_self_review_contamination,
assess_validation_report,
build_final_report,
resolve_repos_from_user_reference,
verify_pinned_head_checkout,
)
@@ -463,5 +466,218 @@ class TestFinalReport(unittest.TestCase):
self.assertFalse(report["merge_allowed"])
class TestStdoutIsolation(unittest.TestCase):
"""Regression test for #178: tests must not close or corrupt stdout/stderr
(prevents need for junitxml workaround in full suite runs and review validation).
"""
def test_stdout_remains_usable(self):
"""After typical test activity (mocks, redirects, prints from mains), stdout should be usable."""
# Verify not closed
self.assertFalse(getattr(sys.stdout, "closed", False))
# Should be able to write (even if captured by pytest)
try:
sys.stdout.write("")
sys.stdout.flush()
except Exception as exc:
self.fail(f"stdout write failed after test activity: {exc}")
def test_stderr_remains_usable(self):
self.assertFalse(getattr(sys.stderr, "closed", False))
try:
sys.stderr.write("")
sys.stderr.flush()
except Exception as exc:
self.fail(f"stderr write failed: {exc}")
class TestRepoNameDisambiguation(unittest.TestCase):
"""Harness assertions for repo name disambiguation (new blind-review hardening).
"MCP Gitea tool" etc. must resolve to Gitea-Tools, not silently default to
mcp-control-plane. "open PRs" or ambiguous must check both. Single-repo
zero-result must not hide PRs in the other configured repo.
"""
CONFIGURED = [
"Scaled-Tech-Consulting/Gitea-Tools",
"Scaled-Tech-Consulting/mcp-control-plane",
]
def test_gitea_tools_aliases_resolve_to_gitea_tools_only(self):
for ref in [
"MCP Gitea tool",
"gitea tool",
"Gitea-Tools",
"gitea mcp tool",
"gitea-tools repo",
]:
result = resolve_repos_from_user_reference(ref, self.CONFIGURED)
self.assertEqual(result, ["Scaled-Tech-Consulting/Gitea-Tools"])
def test_mcp_control_plane_alias_resolves_only_to_it(self):
result = resolve_repos_from_user_reference(
"mcp-control-plane", self.CONFIGURED
)
self.assertEqual(result, ["Scaled-Tech-Consulting/mcp-control-plane"])
def test_ambiguous_or_empty_defaults_to_both(self):
self.assertEqual(
resolve_repos_from_user_reference("open PRs", self.CONFIGURED),
self.CONFIGURED,
)
self.assertEqual(
resolve_repos_from_user_reference("", self.CONFIGURED),
self.CONFIGURED,
)
self.assertEqual(
resolve_repos_from_user_reference("review the queue", self.CONFIGURED),
self.CONFIGURED,
)
def test_single_repo_zero_result_does_not_hide_other(self):
# Simulate a run that only inventoried mcp because of bad alias resolution.
# The inventory must still require Gitea-Tools to claim "no open PRs".
mcp_only_reports = [
{
"repo": "Scaled-Tech-Consulting/mcp-control-plane",
"state_filter": "open",
"pagination_complete": True,
"open_pr_count": 0,
}
]
result = assess_inventory_completeness(
repo_reports=mcp_only_reports,
required_repos=self.CONFIGURED,
)
self.assertFalse(result["complete"])
self.assertTrue(
any("Gitea-Tools" in r for r in result["reasons"])
)
def test_full_inventory_of_both_is_required_for_exhaustive_claim(self):
both_reports = [
{
"repo": "Scaled-Tech-Consulting/Gitea-Tools",
"state_filter": "open",
"pagination_complete": True,
"open_pr_count": 1,
},
{
"repo": "Scaled-Tech-Consulting/mcp-control-plane",
"state_filter": "open",
"pagination_complete": True,
"open_pr_count": 0,
},
]
result = assess_inventory_completeness(
repo_reports=both_reports, required_repos=self.CONFIGURED
)
self.assertTrue(result["complete"])
self.assertTrue(result["can_claim_exhaustive"])
class TestControllerHandoff(unittest.TestCase):
"""Issue #182: final reports must end with a Controller Handoff."""
BASE_HANDOFF = "\n".join([
"## Controller Handoff",
"",
"- Task: implement issue #182",
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
"- Role: author",
"- Identity: jcwalker3 / prgs-author",
"- Issue/PR: #182 / PR #999",
"- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Files changed: review_proofs.py",
"- Validation: 700 passed, 6 skipped",
"- Mutations: one PR opened",
"- Current status: PR open",
"- Blockers: none",
"- Next: review PR #999",
"- Safety: no self-review; no self-merge; no secrets",
])
def test_report_without_handoff_is_downgraded(self):
result = assess_controller_handoff("long report text, no handoff")
self.assertEqual(result["verdict"], "missing")
self.assertTrue(result["downgraded"])
def test_wrong_title_is_downgraded(self):
text = self.BASE_HANDOFF.replace(
"## Controller Handoff", "## Handoff Summary")
result = assess_controller_handoff(text)
self.assertEqual(result["verdict"], "missing")
def test_complete_base_handoff_passes(self):
result = assess_controller_handoff(
"full report body...\n\n" + self.BASE_HANDOFF)
self.assertEqual(result["verdict"], "complete")
self.assertFalse(result["downgraded"])
def test_missing_base_fields_are_listed(self):
text = "\n".join(
line for line in self.BASE_HANDOFF.splitlines()
if not line.startswith(("- Mutations:", "- Safety:")))
result = assess_controller_handoff(text)
self.assertEqual(result["verdict"], "incomplete")
self.assertTrue(result["downgraded"])
self.assertIn("Mutations", result["missing_fields"])
self.assertIn("Safety", result["missing_fields"])
def test_review_role_requires_review_fields(self):
result = assess_controller_handoff(self.BASE_HANDOFF, role="review")
self.assertEqual(result["verdict"], "incomplete")
self.assertIn("Pinned reviewed head", result["missing_fields"])
self.assertIn("Merge result", result["missing_fields"])
complete = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected PR: #999",
"- Reviewer eligibility: passed",
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Review decision: approve",
"- Merge result: merged",
"- Linked issue status: closed",
"- Cleanup status: branch deleted",
])
result = assess_controller_handoff(complete, role="review")
self.assertEqual(result["verdict"], "complete")
def test_author_role_requires_author_fields(self):
complete = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #182",
"- Claim/comment status: comment-claimed",
"- PR number opened: #999",
"- No review/merge: confirmed",
])
result = assess_controller_handoff(complete, role="author")
self.assertEqual(result["verdict"], "complete")
result = assess_controller_handoff(self.BASE_HANDOFF, role="author")
self.assertEqual(result["verdict"], "incomplete")
self.assertIn("No review/merge confirmation", result["missing_fields"])
def test_inventory_role_requires_inventory_fields(self):
complete = self.BASE_HANDOFF + "\n" + "\n".join([
"- Repositories checked: Gitea-Tools, mcp-control-plane",
"- Open PR counts: 2 / 0",
"- Selected PR or reason: none eligible (self-authored)",
"- Inventory completeness: complete, no pagination needed",
])
result = assess_controller_handoff(complete, role="inventory")
self.assertEqual(result["verdict"], "complete")
def test_skill_doc_declares_handoff_requirement(self):
# Doc-contract: SKILL.md must keep requiring the exact section and
# naming this validator, or the convention silently rots.
skill = (
__import__("pathlib").Path(__file__).resolve().parent.parent
/ "skills" / "llm-project-workflow" / "SKILL.md"
).read_text(encoding="utf-8")
self.assertIn("## Controller Handoff", skill)
self.assertIn("assess_controller_handoff", skill)
self.assertIn("issue #182", skill)
if __name__ == "__main__":
unittest.main()