fix(issues): enforce formatter acceptance validation

.github/sync-manifest.yml

@@ -584,6 +584,7 @@ scripts:
     # (format step: `python scripts/langchain/issue_formatter.py`), which does NOT
     # runtime-fetch scripts/langchain. Must stay copy-synced.
     delivery: copy
+    template_sync: exact
 
   - source: scripts/langchain/issue_pr_context.py
     description: "Issue/PR context assembly helper with shared token budget caps"

langsmith-fleet-worker-attempt.json

@@ -1,13 +1,13 @@
 {
   "agent": "codex",
   "cli_version": "0.144.1",
-  "emitted_at": "2026-08-09T02:16:27.270949Z",
+  "emitted_at": "2026-08-09T02:47:58.915158Z",
   "execution_profile": "codex-default",
   "fallback_models": [
     "gpt-5.5"
   ],
   "operation_role": "worker",
-  "pr_number": "2998",
+  "pr_number": "2999",
   "requested_model": "gpt-5.6-terra",
   "runner": "reusable-codex-run",
   "schema": "langsmith-fleet/v1",

scripts/langchain/issue_formatter.py

@@ -10,10 +10,12 @@
 from __future__ import annotations
 
 import argparse
+import importlib.util
 import json
 import os
 import re
 import sys
+from functools import lru_cache
 from pathlib import Path
 from typing import Any
 
@@ -46,6 +48,20 @@
 # ~4 chars per token, so 50k chars ≈ 12.5k tokens, leaving headroom for prompt + output
 MAX_ISSUE_BODY_SIZE = 50000
 
+
+@lru_cache(maxsize=1)
+def _issue_format_validator() -> Any:
+    """Load the fleet's single issue-format definition without forking it."""
+    validator_path = Path(__file__).resolve().parents[2] / ".github/scripts/issue_format.py"
+    spec = importlib.util.spec_from_file_location("_fleet_issue_format", validator_path)
+    if spec is None or spec.loader is None:  # pragma: no cover - repository invariant
+        raise RuntimeError(f"Cannot load canonical issue-format validator: {validator_path}")
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
 # Workflow tags written into the reuse marker. Tagging every stage of the
 # auto-pilot format -> optimize -> apply chain lets any stage detect a body it (or
 # a sibling stage) already formatted and skip re-deriving it, which is the
@@ -151,6 +167,7 @@ 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)
 
 
 def _context_token_budget() -> int:
@@ -355,6 +372,22 @@ def join_or_placeholder(lines: list[str], placeholder: str) -> str:
     impl_text = join_or_placeholder(impl_lines, "_Not provided._")
     tasks_text = join_or_placeholder(tasks_lines, "- [ ] _Not provided._")
     acceptance_text = join_or_placeholder(acceptance_lines, "- [ ] _Not provided._")
+    try:
+        validator = _issue_format_validator()
+    except (ImportError, OSError, RuntimeError, SyntaxError):
+        # 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):
+        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."
+            )
 
     parts = [
         "## Why",
@@ -387,8 +420,12 @@ def join_or_placeholder(lines: list[str], placeholder: str) -> str:
 def _formatted_output_valid(text: str) -> bool:
     if not text:
         return False
-    required = ["## Tasks", "## Acceptance Criteria"]
-    return all(section in text for section in required)
+    try:
+        return bool(_issue_format_validator().validate(text).ok)
+    except (ImportError, OSError, RuntimeError, SyntaxError):
+        # Preserve the former heading-only behavior until the copy-synced
+        # validator becomes available again.
+        return all(section in text for section in ("## Tasks", "## Acceptance Criteria"))
 
 
 def _select_code_fence(text: str) -> str:
@@ -398,12 +435,10 @@ def _select_code_fence(text: str) -> str:
 
 
 ORIGINAL_ISSUE_SUMMARY = "<summary>Original Issue</summary>"
-# Matches an Original-Issue <details> block (and trailing whitespace) so it can
-# be replaced rather than nested. Non-greedy body, anchored to the closing tag.
-_ORIGINAL_ISSUE_BLOCK_RE = re.compile(
-    r"<details>\s*<summary>Original Issue</summary>.*?</details>[ \t]*\n?",
-    re.DOTALL | re.IGNORECASE,
+_ORIGINAL_ISSUE_OPEN_RE = re.compile(
+    r"<details\b[^>]*>\s*<summary>Original Issue</summary>", re.IGNORECASE
 )
+_DETAILS_TAG_RE = re.compile(r"</?details\b[^>]*>", re.IGNORECASE)
 # Captures the verbatim text fenced inside an Original-Issue block, so an
 # already-embedded original can be recovered (and re-embedded once) instead of
 # being wrapped again.
@@ -415,8 +450,25 @@ def _select_code_fence(text: str) -> str:
 
 
 def _strip_original_issue_blocks(text: str) -> str:
-    """Remove any embedded Original-Issue <details> block(s) from ``text``."""
-    return _ORIGINAL_ISSUE_BLOCK_RE.sub("", text).rstrip()
+    """Remove complete embedded Original-Issue blocks, including nested details."""
+    kept: list[str] = []
+    cursor = 0
+    while match := _ORIGINAL_ISSUE_OPEN_RE.search(text, cursor):
+        kept.append(text[cursor : match.start()])
+        depth = 1
+        end = match.end()
+        for tag in _DETAILS_TAG_RE.finditer(text, match.end()):
+            depth += -1 if tag.group(0).startswith("</") else 1
+            if depth == 0:
+                end = tag.end()
+                break
+        else:
+            # Leave malformed markup intact rather than silently discarding it.
+            kept.append(text[match.start() :])
+            return "".join(kept).rstrip()
+        cursor = end
+    kept.append(text[cursor:])
+    return "".join(kept).rstrip()
 
 
 def _innermost_original_issue(text: str) -> str | None:
@@ -705,6 +757,7 @@ 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).
@@ -716,6 +769,7 @@ def format_issue_body(issue_body: str, *, use_llm: bool = True) -> dict[str, Any
         "provider_used": None,
         "used_llm": False,
         "validation_audit": audit,
+        "needs_refinement": needs_refinement,
     }