fix(issues): harden formatter contract handling

.github/scripts/issue_format.py

@@ -92,8 +92,10 @@
     r"LICENSE(?:\.(?:md|txt))?|\.gitignore|\.editorconfig)"
 )
 _TASK_COMMAND = (
-    r"(?:python(?:3)?|pytest|npm|pnpm|yarn|make|just|cargo|go|dotnet|gh|curl|"
-    r"node|vitest|jest|playwright)"
+    r"(?:python(?:3)?\s+-m\s+(?:pytest|unittest)|pytest\b|node\s+--test\b|"
+    r"(?:npm|pnpm|yarn)\s+(?:run\s+)?(?:test|vitest|jest|playwright)\b|"
+    r"(?:make|just|cargo|go|dotnet)\s+(?:test|check)\b|"
+    r"gh\s+(?:workflow\s+run|run)\b|curl\s+\S+)"
 )
 
 
@@ -146,7 +148,9 @@ def _task_has_concrete_target(item: str) -> bool:
     # Unquoted path with a directory separator (src/main.go, .github/workflows/x.yml).
     if re.search(r"(?:^|[\s])((?:\./)?[\w.-]+(?:/[\w./-]+)+)", item):
         return True
-    return re.search(rf"\b{_TASK_COMMAND}\b", item, re.I) is not None
+    # Command names in prose ("make the UI better", "go improve it") are not
+    # concrete targets. Require a command-shaped invocation instead.
+    return re.search(rf"(?:^|\s){_TASK_COMMAND}", item, re.I) is not None
 
 
 def _headings(body: str) -> list[tuple[str, int, int]]:

.github/workflows/agents-auto-pilot.yml

@@ -795,6 +795,12 @@ jobs:
               with open('/tmp/guard_blocked', 'w') as f:
                   f.write(reason)
               sys.exit(0)
+          if result.get('needs_refinement'):
+              reason = 'Formatter output does not satisfy the canonical issue-format contract.'
+              print(f'NEEDS_REFINEMENT: {reason}')
+              with open('/tmp/needs_refinement', 'w') as f:
+                  f.write(reason)
+              sys.exit(0)
           formatted = result.get('formatted_body', '')
           if not formatted:
               print('ERROR: No formatted body returned')
@@ -862,6 +868,33 @@ jobs:
             exit 0
           fi
 
+          # Never publish or mark a body as formatted when the formatter itself
+          # reports that its final output fails the canonical contract.
+          if [ -f /tmp/needs_refinement ]; then
+            echo "⚠️ Formatter requires human refinement; pausing auto-pilot."
+            FORMAT_REFINEMENT_REASON="$(cat /tmp/needs_refinement)" node - <<'REFINEMENT_NODE'
+            (async () => {
+              const { Octokit } = require('@octokit/rest');
+              const { createTokenAwareRetry } = require('./.github/scripts/github-api-with-retry.js');
+              const core = { info: () => {}, warning: console.warn, debug: () => {} };
+              const github = new Octokit({ auth: process.env.GITHUB_TOKEN });
+              const { withRetry } = await createTokenAwareRetry({
+                github, core, env: process.env, task: 'auto-pilot', capabilities: ['issues:write'],
+              });
+              const [owner, repo] = process.env.GITHUB_REPOSITORY.split('/');
+              const issue_number = Number(process.env.ISSUE_NUMBER);
+              await withRetry((client) => client.rest.issues.addLabels({
+                owner, repo, issue_number, labels: ['needs-human', 'agents:auto-pilot-pause'],
+              }));
+              await withRetry((client) => client.rest.issues.createComment({
+                owner, repo, issue_number,
+                body: `⚠️ **Auto-pilot paused during formatting.**\n\n${process.env.FORMAT_REFINEMENT_REASON}`,
+              }));
+            })().catch((error) => { console.error(error); process.exit(1); });
+            REFINEMENT_NODE
+            exit 0
+          fi
+
           # Update issue body
           node - <<'NODE'
           (async () => {

scripts/langchain/issue_formatter.py

@@ -168,6 +168,18 @@ def _strip_reuse_marker(text: str) -> str:
 LIST_ITEM_REGEX = re.compile(r"^(\s*)([-*+]|\d+[.)]|[A-Za-z][.)])\s+(.*)$")
 CHECKBOX_REGEX = re.compile(r"^\[([ xX])\]\s*(.*)$")
 VERIFY_HINT_REGEX = re.compile(r"\(verify:\s*([^\n)]+)\)", re.IGNORECASE)
+SAFE_VERIFY_COMMAND_RE = re.compile(
+    r"^(?:"
+    r"(?:python(?:3)?\s+-m\s+)?pytest\b"
+    r"|node\s+--test\b"
+    r"|(?:npm|pnpm|yarn)\s+(?:run\s+)?(?:test|vitest|jest|playwright)\b"
+    r"|(?:make|just|cargo|go|dotnet)\s+(?:test|check)\b"
+    r"|gh\s+(?:workflow\s+run|run)\b"
+    r"|curl\b"
+    r")",
+    re.IGNORECASE,
+)
+SHELL_METACHARACTERS_RE = re.compile(r"[;&|`$<>\n\r]")
 
 
 def _context_token_budget() -> int:
@@ -378,16 +390,28 @@ def join_or_placeholder(lines: list[str], placeholder: str) -> str:
         # Consumer checkouts can be mid-sync or missing the canonical validator.
         # Keep the pre-validator fallback usable instead of failing the formatter.
         validator = None
-    if validator is not None and not validator.GATE.search(acceptance_text):
+    gate = getattr(validator, "GATE", None) if validator is not None else None
+    if gate is not None and not gate.search(acceptance_text):
         verify_hint = VERIFY_HINT_REGEX.search(tasks_text)
         if verify_hint:
             command = verify_hint.group(1).strip().strip("`")
             if command.startswith("pytest "):
                 command = f"python3 -m {command}"
-            acceptance_text = (
-                f"{acceptance_text}\n"
-                f"- [ ] Run `{command}` and capture the command output in PR validation evidence."
-            )
+            if (
+                SAFE_VERIFY_COMMAND_RE.match(command)
+                and not SHELL_METACHARACTERS_RE.search(command)
+                and gate.search(command)
+            ):
+                if is_placeholder_checklist_text(acceptance_text) or re.fullmatch(
+                    r"- \[ \] _Not provided\._", acceptance_text.strip()
+                ):
+                    acceptance_text = ""
+                criterion = (
+                    f"- [ ] Run `{command}` and capture the command output in PR validation evidence."
+                )
+                acceptance_text = "\n".join(
+                    part for part in (acceptance_text, criterion) if part
+                )
 
     parts = [
         "## Why",
@@ -443,7 +467,7 @@ def _select_code_fence(text: str) -> str:
 # already-embedded original can be recovered (and re-embedded once) instead of
 # being wrapped again.
 _ORIGINAL_ISSUE_INNER_RE = re.compile(
-    r"<details>\s*<summary>Original Issue</summary>\s*"
+    r"<details\b[^>]*>\s*<summary>Original Issue</summary>\s*"
     r"(?P<fence>`{3,})text\n(?P<inner>.*?)\n(?P=fence)\s*</details>",
     re.DOTALL | re.IGNORECASE,
 )
@@ -749,6 +773,7 @@ def format_issue_body(issue_body: str, *, use_llm: bool = True) -> dict[str, Any
                         "provider_used": provider,
                         "used_llm": True,
                         "validation_audit": audit,
+                        "needs_refinement": not _formatted_output_valid(formatted),
                     }
                     result.update(trace.as_dict())
                     return result
@@ -757,13 +782,13 @@ def format_issue_body(issue_body: str, *, use_llm: bool = True) -> dict[str, Any
                 pass
 
     formatted = _format_issue_fallback(issue_body)
-    needs_refinement = not _formatted_output_valid(formatted)
     # NOTE: Task decomposition is now handled by agents:optimize step
     # which uses LLM for intelligent splitting. Don't do heuristic
     # splitting here - it causes task explosion (issue #805, #1143).
     formatted, audit = _validate_and_refine_tasks(formatted, use_llm=use_llm)
     formatted = _append_raw_issue_section(formatted, issue_body)
     formatted = _with_reuse_marker(formatted)
+    needs_refinement = not _formatted_output_valid(formatted)
     return {
         "formatted_body": formatted,
         "provider_used": None,
@@ -809,6 +834,7 @@ def main() -> None:
             "provider_used": result.get("provider_used"),
             "used_llm": result.get("used_llm", False),
             "labels": build_label_transition(),
+            "needs_refinement": result.get("needs_refinement", False),
         }
         if result.get("guard_blocked"):
             payload["guard_blocked"] = True

tests/scripts/test_issue_format.py

@@ -165,6 +165,17 @@ def test_task_without_concrete_target_is_non_conforming() -> None:
     assert "concrete file" in report.as_markdown()
 
 
+@pytest.mark.parametrize("task", ["Make the UI better", "Just fix bugs", "Go improve it"])
+def test_command_words_in_prose_are_not_concrete_targets(task: str) -> None:
+    validator = _validator()
+    report = validator.validate(
+        VALID_CONTEXT
+        + f"## Tasks\n- [ ] {task}\n\n"
+        + "## Acceptance Criteria\n- pytest tests/test_x.py passes\n"
+    )
+    assert not report.ok
+
+
 @pytest.mark.parametrize(
     "task",
     [

tests/scripts/test_issue_formatter.py

@@ -85,6 +85,36 @@ def test_format_issue_fallback_adds_acceptance_gate_when_only_tasks_have_verify_
     assert _canonical_issue_format().validate(formatted).ok is True
 
 
+def test_format_issue_fallback_replaces_acceptance_placeholder_for_safe_verify_hint() -> None:
+    raw = """## Tasks
+- [ ] Update `scripts/langchain/issue_formatter.py` and run `(verify: pytest tests/scripts/test_issue_formatter.py)`.
+
+## Acceptance Criteria
+- [ ] _Not provided._
+"""
+
+    formatted = issue_formatter.format_issue_body(raw, use_llm=False)["formatted_body"]
+    acceptance = _extract_section(formatted, "Acceptance Criteria")
+
+    assert "_Not provided._" not in acceptance
+    assert "python3 -m pytest tests/scripts/test_issue_formatter.py" in acceptance
+
+
+def test_format_issue_fallback_does_not_promote_shell_verify_hint() -> None:
+    raw = """## Tasks
+- [ ] Update `scripts/langchain/issue_formatter.py` `(verify: pytest tests/test_x.py; curl https://example.invalid)`.
+
+## Acceptance Criteria
+- [ ] _Not provided._
+"""
+
+    formatted = issue_formatter.format_issue_body(raw, use_llm=False)["formatted_body"]
+
+    assert "curl https://example.invalid" not in _extract_section(
+        formatted, "Acceptance Criteria"
+    )
+
+
 def test_format_issue_fallback_preserves_tasks_without_decomposition() -> None:
     """Formatter preserves tasks as-is; decomposition is done by agents:optimize step.
 
@@ -587,6 +617,30 @@ def test_append_raw_issue_section_replaces_not_nests() -> None:
     assert "TRUE ORIGINAL" in out_empty_raw
 
 
+def test_append_raw_issue_section_recovers_attributed_details_wrapper() -> None:
+    formatted_with_block = """## Tasks
+
+- [ ] `pytest tests/scripts/test_issue_formatter.py`
+
+## Acceptance Criteria
+
+- [ ] `pytest tests/scripts/test_issue_formatter.py` passes.
+
+<details open>
+<summary>Original Issue</summary>
+
+```text
+TRUE ORIGINAL
+```
+</details>
+"""
+
+    out = issue_formatter._append_raw_issue_section(formatted_with_block, formatted_with_block)
+
+    assert out.count("<summary>Original Issue</summary>") == 1
+    assert "TRUE ORIGINAL" in out
+
+
 def test_append_raw_issue_section_collapses_nested_blocks() -> None:
     """Pre-existing nested blocks collapse to a single block on the next pass."""
     nested = "\n".join(