feat: P2-P5 — HF sync, eval framework, spawner/teams, instinct tests

P2: agent/hf_sync.py — HuggingFace leaderboard sync with model score
  persistence, staleness detection, and benchmark extraction from HF API.

P3: eval/ package — Eval-Driven Development framework with EvalCase,
  EvalResult, EvalSuite, EvalReport types; built-in core suite (5 eval
  cases: code_gen, architecture_review, test_gen, patent_claim,
  intent_classify); stub runner and markdown report generator.

P4: agent/spawner.py — SubAgentSpawner for isolated agent workers with
  role-based tool subsets. agent/teams.py — 4 predefined team
  compositions (pair-review, full-stack, ip-analysis, spec-draft).

P5: Instinct system already existed — added 5 tests for InstinctStore
  (add, remove, persistence, record_accepted, export_markdown).

26 new tests in test_new_modules.py (all passing). Total: 570 tests.

Co-Authored-By: Oz <oz-agent@warp.dev>

Author: tbitcs

src/specsmith/agent/hf_sync.py

@@ -0,0 +1,179 @@
+# SPDX-License-Identifier: MIT
+# Copyright (c) 2026 BitConcepts, LLC. All rights reserved.
+"""HuggingFace Open LLM Leaderboard sync for model intelligence (REQ-223).
+
+Fetches benchmark scores from the HuggingFace API and populates
+`.specsmith/model_scores.json` so that `rank_models_for_role()` uses
+real data instead of hardcoded baselines.
+
+Usage:
+    from specsmith.agent.hf_sync import sync_scores
+    results = sync_scores()  # returns dict of model_id -> {benchmark: score}
+"""
+
+from __future__ import annotations
+
+import json
+import time
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+# HF Inference API endpoint for model info
+HF_API_BASE = "https://huggingface.co/api"
+
+# Models we track (subset of popular models with known benchmark data)
+TRACKED_MODELS: list[str] = [
+    "gpt-4.1",
+    "gpt-4.1-mini",
+    "gpt-4o",
+    "gpt-4o-mini",
+    "claude-sonnet-4-20250514",
+    "claude-3.5-sonnet",
+    "gemini-2.5-pro",
+    "gemini-2.5-flash",
+    "Qwen/Qwen2.5-Coder-32B-Instruct",
+    "Qwen/Qwen2.5-Coder-7B-Instruct",
+    "mistralai/Mistral-Large-Latest",
+    "meta-llama/Llama-3.3-70B-Instruct",
+    "deepseek-ai/DeepSeek-V3",
+]
+
+# Default scores file path (relative to project root)
+SCORES_FILENAME = "model_scores.json"
+
+
+def _scores_path(project_dir: str | Path = ".") -> Path:
+    return Path(project_dir).resolve() / ".specsmith" / SCORES_FILENAME
+
+
+def load_cached_scores(project_dir: str | Path = ".") -> dict[str, Any]:
+    """Load cached model scores from disk."""
+    path = _scores_path(project_dir)
+    if not path.is_file():
+        return {}
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return {}
+
+
+def save_scores(scores: dict[str, Any], project_dir: str | Path = ".") -> None:
+    """Persist model scores to disk."""
+    path = _scores_path(project_dir)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    data = {
+        "synced_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+        "models": scores,
+    }
+    path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
+
+
+def fetch_hf_model_info(model_id: str, timeout: int = 10) -> dict[str, Any]:
+    """Fetch model metadata from HuggingFace API.
+
+    Returns a dict with model card data. On failure returns empty dict.
+    """
+    url = f"{HF_API_BASE}/models/{model_id}"
+    try:
+        req = urllib.request.Request(url, headers={"Accept": "application/json"})
+        with urllib.request.urlopen(req, timeout=timeout) as resp:  # noqa: S310
+            return json.loads(resp.read())
+    except Exception:  # noqa: BLE001
+        return {}
+
+
+def _extract_benchmark_scores(model_info: dict[str, Any]) -> dict[str, float]:
+    """Extract benchmark scores from HF model card metadata.
+
+    Looks for eval_results in the model card data. Returns a dict of
+    benchmark_name -> score.
+    """
+    scores: dict[str, float] = {}
+    # HF model cards store eval results in cardData.eval_results
+    card_data = model_info.get("cardData", {}) or {}
+    eval_results = card_data.get("eval_results", []) or []
+    for result in eval_results:
+        if not isinstance(result, dict):
+            continue
+        dataset = result.get("dataset", {})
+        name = dataset.get("name", "") if isinstance(dataset, dict) else str(dataset)
+        metrics = result.get("metrics", []) or []
+        for metric in metrics:
+            if isinstance(metric, dict):
+                metric_name = metric.get("name", "")
+                value = metric.get("value")
+                if metric_name and value is not None:
+                    try:
+                        key = f"{name}/{metric_name}" if name else metric_name
+                        scores[key] = float(value)
+                    except (TypeError, ValueError):
+                        continue
+    return scores
+
+
+def sync_scores(
+    project_dir: str | Path = ".",
+    models: list[str] | None = None,
+    timeout: int = 10,
+) -> dict[str, dict[str, float]]:
+    """Sync model scores from HuggingFace.
+
+    For HF-hosted models, fetches real benchmark data from model cards.
+    For proprietary models (GPT, Claude, Gemini), uses curated baselines.
+
+    Returns dict of model_id -> {benchmark: score}.
+    """
+    from specsmith.agent.model_intelligence import BASELINE_SCORES
+
+    target_models = models or TRACKED_MODELS
+    all_scores: dict[str, dict[str, float]] = {}
+
+    for model_id in target_models:
+        # For non-HF models, use baseline scores
+        if "/" not in model_id:
+            baseline = BASELINE_SCORES.get(model_id)
+            if baseline:
+                all_scores[model_id] = {"baseline_composite": baseline}
+            continue
+
+        # For HF models, try to fetch real data
+        info = fetch_hf_model_info(model_id, timeout=timeout)
+        if info:
+            benchmarks = _extract_benchmark_scores(info)
+            if benchmarks:
+                all_scores[model_id] = benchmarks
+                continue
+
+        # Fallback to baseline
+        baseline = BASELINE_SCORES.get(model_id)
+        if baseline:
+            all_scores[model_id] = {"baseline_composite": baseline}
+
+    save_scores(all_scores, project_dir)
+    return all_scores
+
+
+def is_stale(project_dir: str | Path = ".", max_age_hours: int = 24) -> bool:
+    """Check if cached scores are older than max_age_hours."""
+    cached = load_cached_scores(project_dir)
+    synced_at = cached.get("synced_at", "")
+    if not synced_at:
+        return True
+    try:
+        from datetime import datetime, timezone
+
+        synced = datetime.fromisoformat(synced_at.replace("Z", "+00:00"))
+        age = datetime.now(timezone.utc) - synced
+        return age.total_seconds() > max_age_hours * 3600
+    except (ValueError, TypeError):
+        return True
+
+
+__all__ = [
+    "fetch_hf_model_info",
+    "is_stale",
+    "load_cached_scores",
+    "save_scores",
+    "sync_scores",
+]

src/specsmith/agent/spawner.py

@@ -0,0 +1,92 @@
+# SPDX-License-Identifier: MIT
+# Copyright (c) 2026 BitConcepts, LLC. All rights reserved.
+"""Sub-agent spawner — spawn isolated agent workers with tool subsets.
+
+ARCHITECTURE.md §13 Phase 2: Multi-Agent Layer.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class SpawnedAgent:
+    """Metadata for a spawned sub-agent."""
+
+    id: str
+    role: str
+    tools: list[str]
+    status: str = "idle"  # idle, running, completed, failed
+    result: dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "id": self.id,
+            "role": self.role,
+            "tools": self.tools,
+            "status": self.status,
+            "result": self.result,
+        }
+
+
+# Tool subsets for different agent roles
+ROLE_TOOLS: dict[str, list[str]] = {
+    "coder": ["read_file", "write_file", "run_shell", "apply_diff"],
+    "reviewer": ["read_file", "run_shell", "git_diff"],
+    "tester": ["read_file", "run_shell", "run_tests"],
+    "architect": ["read_file", "write_file"],
+    "researcher": ["read_file", "search_web", "search_repo"],
+}
+
+
+class SubAgentSpawner:
+    """Spawn and manage isolated agent workers.
+
+    Each spawned agent gets a restricted tool subset based on its role,
+    preventing accidental cross-domain actions (e.g., a reviewer can't
+    write files).
+    """
+
+    def __init__(self) -> None:
+        self._agents: dict[str, SpawnedAgent] = {}
+        self._counter = 0
+
+    def spawn(self, role: str, tools: list[str] | None = None) -> SpawnedAgent:
+        """Spawn a new sub-agent with the given role and tool set."""
+        self._counter += 1
+        agent_id = f"agent-{role}-{self._counter:03d}"
+        effective_tools = tools or ROLE_TOOLS.get(role, [])
+        agent = SpawnedAgent(id=agent_id, role=role, tools=effective_tools)
+        self._agents[agent_id] = agent
+        return agent
+
+    def get(self, agent_id: str) -> SpawnedAgent | None:
+        """Get a spawned agent by ID."""
+        return self._agents.get(agent_id)
+
+    def list_active(self) -> list[SpawnedAgent]:
+        """List all agents that are not completed/failed."""
+        return [a for a in self._agents.values() if a.status in ("idle", "running")]
+
+    def list_all(self) -> list[SpawnedAgent]:
+        """List all spawned agents."""
+        return list(self._agents.values())
+
+    def complete(self, agent_id: str, result: dict[str, Any]) -> None:
+        """Mark an agent as completed with its result."""
+        agent = self._agents.get(agent_id)
+        if agent:
+            agent.status = "completed"
+            agent.result = result
+
+    def fail(self, agent_id: str, error: str) -> None:
+        """Mark an agent as failed."""
+        agent = self._agents.get(agent_id)
+        if agent:
+            agent.status = "failed"
+            agent.result = {"error": error}
+
+
+__all__ = ["ROLE_TOOLS", "SpawnedAgent", "SubAgentSpawner"]

src/specsmith/agent/teams.py

@@ -0,0 +1,103 @@
+# SPDX-License-Identifier: MIT
+# Copyright (c) 2026 BitConcepts, LLC. All rights reserved.
+"""Team definitions for multi-agent coordination.
+
+ARCHITECTURE.md §13 Phase 2: predefined agent team compositions.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class TeamMember:
+    """A role slot within a team."""
+
+    role: str
+    required: bool = True
+    tools_override: list[str] | None = None
+
+
+@dataclass
+class TeamDefinition:
+    """A named team of agent roles that work together."""
+
+    id: str
+    name: str
+    description: str
+    members: list[TeamMember] = field(default_factory=list)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "id": self.id,
+            "name": self.name,
+            "description": self.description,
+            "members": [{"role": m.role, "required": m.required} for m in self.members],
+        }
+
+
+# Pre-defined teams
+PAIR_REVIEW = TeamDefinition(
+    id="pair-review",
+    name="Pair Review",
+    description="Coder + Reviewer pair for code changes with built-in review",
+    members=[
+        TeamMember(role="coder"),
+        TeamMember(role="reviewer"),
+    ],
+)
+
+FULL_STACK = TeamDefinition(
+    id="full-stack",
+    name="Full Stack",
+    description="Architect + Coder + Tester trio for complete feature development",
+    members=[
+        TeamMember(role="architect"),
+        TeamMember(role="coder"),
+        TeamMember(role="tester"),
+    ],
+)
+
+IP_ANALYSIS = TeamDefinition(
+    id="ip-analysis",
+    name="IP Analysis",
+    description="IP Analyst + Researcher + Strategist for patent work",
+    members=[
+        TeamMember(role="ip-analyst"),
+        TeamMember(role="researcher"),
+        TeamMember(role="strategist"),
+    ],
+)
+
+SPEC_DRAFT = TeamDefinition(
+    id="spec-draft",
+    name="Specification Drafting",
+    description="Architect + Drafter + Reviewer for specification writing",
+    members=[
+        TeamMember(role="architect"),
+        TeamMember(role="drafter"),
+        TeamMember(role="reviewer"),
+    ],
+)
+
+BUILTIN_TEAMS: dict[str, TeamDefinition] = {
+    "pair-review": PAIR_REVIEW,
+    "full-stack": FULL_STACK,
+    "ip-analysis": IP_ANALYSIS,
+    "spec-draft": SPEC_DRAFT,
+}
+
+
+def get_team(team_id: str) -> TeamDefinition | None:
+    """Get a built-in team by ID."""
+    return BUILTIN_TEAMS.get(team_id)
+
+
+def list_teams() -> list[TeamDefinition]:
+    """List all available teams."""
+    return list(BUILTIN_TEAMS.values())
+
+
+__all__ = ["BUILTIN_TEAMS", "TeamDefinition", "TeamMember", "get_team", "list_teams"]