Implement Hardware Support Agent Specification and Workflow

bin/board_intake.py

@@ -0,0 +1,628 @@
+#!/usr/bin/env python3
+"""Board intake assessment for new Meshtastic hardware support.
+
+Validates a board intake request against the repository hardware context,
+identifies evidence gaps, and produces a structured readiness report.
+
+Usage:
+    python3 bin/board_intake.py <intake.json>
+    python3 bin/board_intake.py <intake.json> --output report.md
+    python3 bin/board_intake.py <intake.json> --validate   # gaps check only, exit 1 if not scaffold_ready
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sys
+from dataclasses import dataclass, field
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_CONTEXT_PATH = ROOT / "docs" / "hardware-support-context.md"
+
+# Metadata keys every new PlatformIO environment should declare.
+REQUIRED_METADATA_KEYS = [
+    "custom_meshtastic_hw_model",
+    "custom_meshtastic_hw_model_slug",
+    "custom_meshtastic_architecture",
+    "custom_meshtastic_actively_supported",
+    "custom_meshtastic_support_level",
+    "custom_meshtastic_display_name",
+]
+
+RECOMMENDED_METADATA_KEYS = [
+    "custom_meshtastic_images",
+    "custom_meshtastic_tags",
+    "custom_meshtastic_requires_dfu",
+    "custom_meshtastic_partition_scheme",
+]
+
+# Pin groups that should be backed by evidence before scaffolding.
+EVIDENCE_CATEGORIES = ["radio", "display", "input", "GPS", "power"]
+
+# Architecture families that rely on BSP defaults for many pin defines.
+BSP_DEFAULT_FAMILIES = {"nrf52840", "rp2040", "stm32", "native"}
+
+# Known valid architectures from the repository.
+KNOWN_ARCHITECTURES = {
+    "esp32",
+    "esp32-s3",
+    "esp32-c3",
+    "esp32-c6",
+    "esp32s2",
+    "nrf52840",
+    "rp2040",
+    "rp2350",
+    "stm32",
+    "native",
+}
+
+
+# ---------------------------------------------------------------------------
+# T004 – BoardIntakeRequest dataclass
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class BoardIntakeRequest:
+    """Maintainer-supplied description of a proposed new board."""
+
+    # Required
+    environment_name: str
+    hardware_model: str
+    display_name: str
+    architecture: str
+
+    # Recommended
+    hardware_model_slug: str = ""
+    actively_supported: bool | None = None
+    support_level: str = ""
+    source_materials: list[str] = field(default_factory=list)
+    board_notes: str = ""
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "BoardIntakeRequest":
+        return cls(
+            environment_name=data.get("environment_name", ""),
+            hardware_model=str(data.get("hardware_model", "")),
+            display_name=data.get("display_name", ""),
+            architecture=data.get("architecture", ""),
+            hardware_model_slug=data.get("hardware_model_slug", ""),
+            actively_supported=data.get("actively_supported"),
+            support_level=str(data.get("support_level", "")),
+            source_materials=list(data.get("source_materials", [])),
+            board_notes=data.get("board_notes", ""),
+        )
+
+    @classmethod
+    def from_json(cls, path: Path) -> "BoardIntakeRequest":
+        data = json.loads(path.read_text(encoding="utf-8"))
+        return cls.from_dict(data)
+
+
+# ---------------------------------------------------------------------------
+# T005 – EvidenceGap dataclass
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class EvidenceGap:
+    """A specific missing, conflicting, or ambiguous hardware fact."""
+
+    category: str  # metadata | radio | display | input | GPS | power | storage | connectivity | revision-scope
+    description: str
+    affected_artifact: str
+    required_evidence: str
+    blocking: bool
+
+
+# ---------------------------------------------------------------------------
+# T006 – IntakeAssessment dataclass
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class IntakeAssessment:
+    """Structured result of evaluating a BoardIntakeRequest."""
+
+    request: BoardIntakeRequest
+    expected_artifacts: list[str]
+    required_metadata: list[str]
+    matched_patterns: list[dict]
+    evidence_gaps: list[EvidenceGap]
+    risk_flags: list[str]
+    next_actions: list[str]
+    scaffold_ready: bool
+
+
+# ---------------------------------------------------------------------------
+# T007 – load_hardware_context
+# ---------------------------------------------------------------------------
+
+
+def load_hardware_context(path: Path = DEFAULT_CONTEXT_PATH) -> dict:
+    """Parse the generated hardware-support-context.md into a usable dict.
+
+    Returns:
+        {
+            "architecture_names": list[str],  # families present in the inventory
+            "metadata_keys": list[str],       # custom_meshtastic_* keys observed
+            "environments": dict[str, dict],  # env_name -> {display, hw_model, hw_slug, variant_dir, arch}
+        }
+    """
+    if not path.exists():
+        raise FileNotFoundError(
+            f"Hardware context not found at {path}. "
+            "Run: python3 bin/generate_hardware_support_context.py"
+        )
+
+    text = path.read_text(encoding="utf-8")
+
+    # Extract metadata keys from the "## Repository Metadata Inputs" section.
+    metadata_keys: list[str] = re.findall(r"`(custom_meshtastic_[^`]+)`", text)
+    metadata_keys = list(dict.fromkeys(metadata_keys))  # deduplicate, preserve order
+
+    # Extract architecture families from the "## Architecture and Environment Inventory" section.
+    arch_names: list[str] = re.findall(r"^### ([a-z0-9\-]+)\s*$", text, re.MULTILINE)
+    # Filter out sub-headings that are architecture names (exclude e.g. "nrf52840" inside examples)
+    # The inventory section has short arch names; filter to known set plus any that look like archs.
+    arch_names = [a for a in dict.fromkeys(arch_names) if not a[0].isupper()]
+
+    # Extract environments from inventory table rows.
+    environments: dict[str, dict] = {}
+    current_arch = ""
+    for line in text.splitlines():
+        arch_match = re.match(r"^### ([a-z0-9\-]+)\s*$", line)
+        if arch_match:
+            current_arch = arch_match.group(1)
+            continue
+        # Table row: | env | display | hw_model | hw_slug | variant_dir | categories |
+        row = re.match(
+            r"^\|\s*([^|]+?)\s*\|\s*([^|]*?)\s*\|\s*([^|]*?)\s*\|\s*([^|]*?)\s*\|\s*([^|]*?)\s*\|",
+            line,
+        )
+        if (
+            row
+            and not row.group(1).startswith("Environment")
+            and not row.group(1).startswith("---")
+        ):
+            env_name = row.group(1).strip()
+            if env_name:
+                environments[env_name] = {
+                    "display_name": row.group(2).strip(),
+                    "hw_model": row.group(3).strip(),
+                    "hw_slug": row.group(4).strip(),
+                    "variant_dir": row.group(5).strip(),
+                    "architecture": current_arch,
+                }
+
+    return {
+        "architecture_names": arch_names,
+        "metadata_keys": metadata_keys,
+        "environments": environments,
+    }
+
+
+# ---------------------------------------------------------------------------
+# T018 – validate_intake
+# ---------------------------------------------------------------------------
+
+
+def validate_intake(request: BoardIntakeRequest, context: dict) -> list[str]:
+    """Check required fields and detect conflicts with existing environments/models.
+
+    Returns a list of validation error strings (empty = valid).
+    """
+    errors: list[str] = []
+
+    if not request.environment_name:
+        errors.append("environment_name is required.")
+    if not request.hardware_model:
+        errors.append("hardware_model is required.")
+    if not request.display_name:
+        errors.append("display_name is required.")
+    if not request.architecture:
+        errors.append("architecture is required.")
+
+    if request.architecture and request.architecture not in KNOWN_ARCHITECTURES:
+        errors.append(
+            f"architecture '{request.architecture}' is not a known repository architecture. "
+            f"Known: {', '.join(sorted(KNOWN_ARCHITECTURES))}"
+        )
+
+    # Conflict: environment name already exists
+    if request.environment_name and request.environment_name in context.get(
+        "environments", {}
+    ):
+        errors.append(
+            f"environment_name '{request.environment_name}' already exists in the repository. "
+            "Choose a unique name or confirm this is an intentional update."
+        )
+
+    # Conflict: hardware model already assigned to a different environment
+    if request.hardware_model:
+        for env_name, env_data in context.get("environments", {}).items():
+            if (
+                env_data.get("hw_model") == request.hardware_model
+                and env_name != request.environment_name
+            ):
+                errors.append(
+                    f"hardware_model '{request.hardware_model}' is already assigned to "
+                    f"environment '{env_name}' ({env_data.get('display_name', '')!r}). "
+                    "Verify this is a new model number or confirm the shared-model intent."
+                )
+                break  # report once
+
+    return errors
+
+
+# ---------------------------------------------------------------------------
+# T019 – find_matched_patterns
+# ---------------------------------------------------------------------------
+
+
+def find_matched_patterns(request: BoardIntakeRequest, context: dict) -> list[dict]:
+    """Return up to 3 existing environments closest to the request by architecture."""
+    envs = context.get("environments", {})
+    arch = request.architecture
+
+    # Prefer exact architecture match, then partial (e.g., "esp32" matches "esp32-s3").
+    exact: list[dict] = []
+    partial: list[dict] = []
+    for env_name, env_data in envs.items():
+        entry = {**env_data, "environment": env_name}
+        env_arch = env_data.get("architecture", "")
+        if env_arch == arch:
+            exact.append(entry)
+        elif arch and (env_arch.startswith(arch) or arch.startswith(env_arch)):
+            partial.append(entry)
+
+    candidates = exact + partial
+    # Prefer boards that have a display_name and hw_model (more complete entries).
+    candidates.sort(key=lambda e: (not e.get("display_name"), not e.get("hw_model")))
+    return candidates[:3]
+
+
+# ---------------------------------------------------------------------------
+# T020 – build_evidence_gaps
+# ---------------------------------------------------------------------------
+
+
+def build_evidence_gaps(request: BoardIntakeRequest) -> list[EvidenceGap]:
+    """Identify missing pin-group evidence and metadata gaps."""
+    gaps: list[EvidenceGap] = []
+    has_sources = bool(request.source_materials)
+
+    if not has_sources:
+        # Every pin category is unresolvable without sources.
+        for cat in EVIDENCE_CATEGORIES:
+            gaps.append(
+                EvidenceGap(
+                    category=cat,
+                    description=(
+                        f"No source materials supplied. {cat.capitalize()} pin mappings cannot be "
+                        "verified without a schematic, pinout diagram, or vendor datasheet."
+                    ),
+                    affected_artifact="variant.h",
+                    required_evidence="Schematic, pinout image, or vendor board page",
+                    blocking=True,
+                )
+            )
+    else:
+        # Sources exist but may still be incomplete; flag as non-blocking advisory.
+        for cat in EVIDENCE_CATEGORIES:
+            gaps.append(
+                EvidenceGap(
+                    category=cat,
+                    description=(
+                        f"Source materials are present but {cat} pin assignments have not been "
+                        "extracted and cross-checked against repository macro conventions."
+                    ),
+                    affected_artifact="variant.h",
+                    required_evidence=f"Explicit {cat} pin listing matched to repository #define names",
+                    blocking=False,
+                )
+            )
+
+    # Metadata gaps
+    if not request.hardware_model_slug:
+        gaps.append(
+            EvidenceGap(
+                category="metadata",
+                description="hardware_model_slug is not set. The repository requires an UPPER_SNAKE_CASE slug for PlatformIO metadata.",
+                affected_artifact="platformio.ini (custom_meshtastic_hw_model_slug)",
+                required_evidence="Agreed slug from the project maintainers",
+                blocking=True,
+            )
+        )
+
+    if request.actively_supported is None:
+        gaps.append(
+            EvidenceGap(
+                category="metadata",
+                description="actively_supported is not specified. This controls CI matrix inclusion.",
+                affected_artifact="platformio.ini (custom_meshtastic_actively_supported)",
+                required_evidence="Maintainer decision on support status",
+                blocking=False,
+            )
+        )
+
+    if not request.support_level:
+        gaps.append(
+            EvidenceGap(
+                category="metadata",
+                description="support_level is not specified (expected: 1 = active, 2 = supported, 3 = extra).",
+                affected_artifact="platformio.ini (custom_meshtastic_support_level)",
+                required_evidence="Maintainer decision on support tier",
+                blocking=False,
+            )
+        )
+
+    # Revision-scope: multiple display/radio options without disambiguation.
+    # Match whole-word choice language rather than raw substrings so normal text
+    # like "radio" does not trigger the ambiguity gate.
+    note_text = request.board_notes.lower()
+    revision_scope_patterns = (
+        r"\brevision\b",
+        r"\bvariant\b",
+        r"\bvariants\b",
+        r"\boption\b",
+        r"\boptions\b",
+        r"\balternative\b",
+        r"\balternatives\b",
+        r"\bmulti\b",
+        r"\btwo\b",
+        r"\beither\b",
+        r"\bor\b",
+    )
+    if note_text and any(re.search(pattern, note_text) for pattern in revision_scope_patterns):
+        gaps.append(
+            EvidenceGap(
+                category="revision-scope",
+                description=(
+                    "Board notes mention multiple variants, revisions, or options. "
+                    "The intake must be scoped to a single hardware revision before scaffolding can proceed."
+                ),
+                affected_artifact="variant.h, platformio.ini",
+                required_evidence="Explicit decision on which revision this intake covers",
+                blocking=True,
+            )
+        )
+
+    return gaps
+
+
+# ---------------------------------------------------------------------------
+# T021 – assess_intake
+# ---------------------------------------------------------------------------
+
+
+def assess_intake(request: BoardIntakeRequest, context: dict) -> IntakeAssessment:
+    """Produce a full structured assessment from intake request and context."""
+    validation_errors = validate_intake(request, context)
+    matched = find_matched_patterns(request, context)
+    gaps = build_evidence_gaps(request)
+
+    expected_artifacts = [
+        f"variants/{request.architecture or '<architecture>'}/<variant-dir>/variant.h",
+        f"variants/{request.architecture or '<architecture>'}/<variant-dir>/platformio.ini (env:{request.environment_name or '<env>'})",
+    ]
+    if request.architecture in {"esp32", "esp32-s3", "esp32-c3", "esp32-c6"}:
+        expected_artifacts.append(
+            "(optional) variants/.../variant.cpp — only if board requires custom init hooks"
+        )
+    expected_artifacts += [
+        "PlatformIO metadata: all required custom_meshtastic_* keys (see required_metadata below)",
+        "(optional) board image under branding/ or images/ if custom_meshtastic_images is set",
+    ]
+
+    required_metadata = list(REQUIRED_METADATA_KEYS)
+    if request.architecture in BSP_DEFAULT_FAMILIES:
+        required_metadata.append(
+            f"(BSP note) {request.architecture} boards may inherit some pin defines from BSP headers — "
+            "check the Inherited Defaults section of docs/hardware-support-context.md before assuming a missing define is an error."
+        )
+
+    risk_flags: list[str] = []
+    for err in validation_errors:
+        risk_flags.append(f"Validation error: {err}")
+    if request.architecture in BSP_DEFAULT_FAMILIES:
+        risk_flags.append(
+            f"Architecture '{request.architecture}' uses BSP defaults for some pin defines. "
+            "Verify which macros are inherited before declaring them explicitly in variant.h."
+        )
+    if not matched:
+        risk_flags.append(
+            "No closely matched existing board found for this architecture. "
+            "Manual review of variant structure is required."
+        )
+
+    next_actions: list[str] = []
+    if validation_errors:
+        next_actions.append("Resolve validation errors before proceeding.")
+    blocking_gaps = [g for g in gaps if g.blocking]
+    non_blocking_gaps = [g for g in gaps if not g.blocking]
+    for gap in blocking_gaps:
+        next_actions.append(
+            f"Provide {gap.required_evidence} for {gap.category} ({gap.affected_artifact})."
+        )
+    if non_blocking_gaps:
+        next_actions.append(
+            f"Review {len(non_blocking_gaps)} non-blocking gap(s) before merging scaffold output."
+        )
+    if not next_actions:
+        next_actions.append(
+            "All required evidence is present. Proceed to scaffold generation."
+        )
+
+    scaffold_ready = len(validation_errors) == 0 and all(not g.blocking for g in gaps)
+
+    return IntakeAssessment(
+        request=request,
+        expected_artifacts=expected_artifacts,
+        required_metadata=required_metadata,
+        matched_patterns=matched,
+        evidence_gaps=gaps,
+        risk_flags=risk_flags,
+        next_actions=next_actions,
+        scaffold_ready=scaffold_ready,
+    )
+
+
+# ---------------------------------------------------------------------------
+# T022 – render_assessment_markdown
+# ---------------------------------------------------------------------------
+
+
+def render_assessment_markdown(assessment: IntakeAssessment) -> str:
+    req = assessment.request
+    lines: list[str] = []
+
+    lines.append(f"# Board Intake Assessment: `{req.environment_name or '(unnamed)'}`")
+    lines.append("")
+    lines.append(
+        f"**Scaffold ready**: {'✅ Yes' if assessment.scaffold_ready else '❌ No — see blocking gaps below'}"
+    )
+    lines.append("")
+
+    lines.append("## Request Summary")
+    lines.append("")
+    lines.append(f"- **Environment name**: `{req.environment_name}`")
+    lines.append(f"- **Hardware model**: `{req.hardware_model}`")
+    lines.append(
+        f"- **Hardware model slug**: `{req.hardware_model_slug or '(not set)'}`"
+    )
+    lines.append(f"- **Display name**: {req.display_name}")
+    lines.append(f"- **Architecture**: `{req.architecture}`")
+    if req.actively_supported is not None:
+        lines.append(f"- **Actively supported**: {req.actively_supported}")
+    if req.support_level:
+        lines.append(f"- **Support level**: {req.support_level}")
+    if req.source_materials:
+        lines.append("- **Source materials**:")
+        for src in req.source_materials:
+            lines.append(f"  - {src}")
+    if req.board_notes:
+        lines.append(f"- **Board notes**: {req.board_notes}")
+    lines.append("")
+
+    lines.append("## Expected Artifacts")
+    lines.append("")
+    for artifact in assessment.expected_artifacts:
+        lines.append(f"- {artifact}")
+    lines.append("")
+
+    lines.append("## Required Metadata")
+    lines.append("")
+    for key in assessment.required_metadata:
+        lines.append(
+            f"- `{key}`" if key.startswith("custom_meshtastic") else f"- {key}"
+        )
+    lines.append("")
+
+    lines.append("## Matched Repository Patterns")
+    lines.append("")
+    if assessment.matched_patterns:
+        for pattern in assessment.matched_patterns:
+            name = pattern.get("display_name") or pattern.get("environment", "")
+            env = pattern.get("environment", "")
+            arch = pattern.get("architecture", "")
+            vdir = pattern.get("variant_dir", "")
+            lines.append(f"- **{name}** (`{env}`, {arch}) — `{vdir}`")
+    else:
+        lines.append("- No closely matched patterns found for this architecture.")
+    lines.append("")
+
+    blocking = [g for g in assessment.evidence_gaps if g.blocking]
+    non_blocking = [g for g in assessment.evidence_gaps if not g.blocking]
+
+    lines.append("## Evidence Gaps")
+    lines.append("")
+    if blocking:
+        lines.append("### Blocking")
+        lines.append("")
+        for gap in blocking:
+            lines.append(f"- **[{gap.category}]** {gap.description}")
+            lines.append(f"  - Affected: `{gap.affected_artifact}`")
+            lines.append(f"  - Required evidence: {gap.required_evidence}")
+    else:
+        lines.append("*No blocking evidence gaps.*")
+    lines.append("")
+    if non_blocking:
+        lines.append("### Non-blocking (review before merge)")
+        lines.append("")
+        for gap in non_blocking:
+            lines.append(f"- **[{gap.category}]** {gap.description}")
+            lines.append(f"  - Affected: `{gap.affected_artifact}`")
+            lines.append(f"  - Required evidence: {gap.required_evidence}")
+        lines.append("")
+
+    if assessment.risk_flags:
+        lines.append("## Risk Flags")
+        lines.append("")
+        for flag in assessment.risk_flags:
+            lines.append(f"- {flag}")
+        lines.append("")
+
+    lines.append("## Next Actions")
+    lines.append("")
+    for i, action in enumerate(assessment.next_actions, 1):
+        lines.append(f"{i}. {action}")
+    lines.append("")
+
+    return "\n".join(lines)
+
+
+# ---------------------------------------------------------------------------
+# T023 – CLI entry point
+# ---------------------------------------------------------------------------
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Evaluate a board intake request against the repository hardware context."
+    )
+    parser.add_argument("intake", help="Path to the intake JSON file")
+    parser.add_argument(
+        "--output",
+        default="-",
+        help="Output path for the assessment markdown (default: stdout)",
+    )
+    parser.add_argument(
+        "--validate",
+        action="store_true",
+        help="Exit with code 1 if scaffold_ready is false (useful for CI gate checks)",
+    )
+    parser.add_argument(
+        "--context",
+        default=str(DEFAULT_CONTEXT_PATH),
+        help="Path to docs/hardware-support-context.md (default: auto-detected)",
+    )
+    args = parser.parse_args()
+
+    intake_path = Path(args.intake)
+    if not intake_path.exists():
+        print(f"Error: intake file not found: {intake_path}", file=sys.stderr)
+        sys.exit(2)
+
+    request = BoardIntakeRequest.from_json(intake_path)
+    context = load_hardware_context(Path(args.context))
+    assessment = assess_intake(request, context)
+    report = render_assessment_markdown(assessment)
+
+    if args.output == "-":
+        print(report)
+    else:
+        output_path = Path(args.output)
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        output_path.write_text(report, encoding="utf-8")
+        print(f"Assessment written to {output_path}")
+
+    if args.validate and not assessment.scaffold_ready:
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()