Build a source

metalworks sources scaffold mysource --lane grounding --auth none

from collections.abc import Iterator, Sequence

from metalworks.contract import CorpusComment, CorpusRecord
from metalworks.research.sources import SourceSpec, SourceWindow, register_source


class StackExchangeSource:
    """An ItemSource over the public, keyless Stack Exchange API."""

    source_id = "stackexchange"

    def pull(
        self, *, query: str, window: SourceWindow, limit: int | None = None
    ) -> Iterator[CorpusRecord]:
        # Page the search API for `query` over `window`, mapping each question
        # onto a CorpusRecord with a STABLE id, its permalink (url), a
        # pseudonymized author_hash, and a `signals` vector.
        for item in self._search(query, window):
            yield CorpusRecord(
                id=str(item["question_id"]),         # stable: the corpus upserts by id
                source="stackexchange",
                source_id=str(item["question_id"]),
                url=item["link"],                    # the real permalink — cite-or-die
                title=item["title"],
                text=clean(item["body"]),
                author_hash=hash_author(item["owner"]),  # pseudonymized HERE
                engagement=item["score"],
                signals={"votes": float(item["score"]), "views": float(item["view_count"])},
                created_at=to_dt(item["creation_date"]),
            )

    def comments_for(
        self, record_ids: Sequence[str]
    ) -> Iterator[list[CorpusComment]] | None:
        # Yield one CorpusComment batch per record id, in input order. Return
        # None ONLY if your source has no comment layer at all.
        return self._answers_for(record_ids)

    def latest_window(self) -> SourceWindow:
        # The most recent window your source can serve (its anchor).
        ...

register_source(
    "stackexchange",
    lambda **_: StackExchangeSource(),
    spec=SourceSpec(
        source_id="stackexchange",
        lane="grounding",
        signals=("votes", "views"),   # votes = social; views = magnitude
        targeting="instance",         # the selector varies on the SE site
        auth="none",
        env=(),
        access="open",
        relevance_hint=(
            "developers, sysadmins, DBAs, security & cloud/SaaS pros across "
            "170+ Stack Exchange sites"
        ),
    ),
)

BUILTIN_SOURCE_MODULES: dict[str, str] = {
    ...
    "stackexchange": "metalworks.research.sources.stackexchange",
    "mysource": "metalworks.research.sources.mysource",   # ← add one line
}

from metalworks.research.synthesis.signals import SignalSpec, register_signal

register_signal(SignalSpec(kind="likes", weight=1.0, transform="log"))

class MySource:
    source_id = "mysource"
    yields_units = True   # records are the unit; rank by distinct DOMAIN, not author

from metalworks.testing import check_item_source
check_item_source(StackExchangeSource())

from collections.abc import Sequence
from typing import Protocol

class MagnitudeProvider(Protocol):
    provider_id: str
    signals: tuple[str, ...]          # the is_magnitude kinds it emits, e.g. ("downloads",)

    def measure(
        self, *, entities: Sequence[str], window: SourceWindow
    ) -> dict[str, dict[str, float]]:
        """Measure `entities` over `window` → entity -> {kind: value}."""
        ...

from dataclasses import dataclass
from collections.abc import Sequence


@dataclass
class NpmDownloadsProvider:
    provider_id: str = "npm"
    signals: tuple[str, ...] = ("downloads",)

    def measure(
        self, *, entities: Sequence[str], window: SourceWindow | None = None
    ) -> dict[str, dict[str, float]]:
        out: dict[str, dict[str, float]] = {}
        for entity in entities:
            if not _is_npm_package_name(entity):
                continue                       # not package-shaped → skip, never mis-query
            count = self._fetch_downloads(entity, window)
            if count is not None:              # omission = unknown; NEVER store 0.0
                out[entity] = {"downloads": float(count)}
        return out

from metalworks.research.sources.magnitude import (
    MagnitudeSpec,
    register_magnitude,
)

register_magnitude(
    "npm",
    lambda **kwargs: NpmDownloadsProvider(**kwargs),
    spec=MagnitudeSpec(
        provider_id="npm",
        signals=("downloads",),
        targeting="slug",
        auth="none",
        env=(),
        access="open",
        relevance_hint="package adoption / install demand for a named npm package",
    ),
)

DiscoveryProvider (agentic):  Exa Research / Parallel Task → DELEGATE (homegrown loop OFF)
   else → HomegrownDiscovery   (metalworks' own iterate-and-dig loop over a SearchProvider)
      else → web.py single-pass _external_search   (unchanged fallback)

from typing import ClassVar, Protocol

class DiscoveryProvider(Protocol):
    protocol_version: ClassVar[str]
    provider_id: str
    agentic: bool                      # the gate signal — see below

    def discover(
        self, *, question: str, directions: list[str], budget: DiscoveryBudget
    ) -> list[DiscoveryFinding]: ...

class ExaResearchDiscovery:
    protocol_version: ClassVar[str] = PROTOCOL_VERSION
    provider_id: str = "exa_research"
    agentic: bool = True              # → metalworks delegates the whole loop

    def discover(self, *, question, directions, budget) -> list[DiscoveryFinding]:
        task = self._run_research(_instructions(question, directions))
        if task is None:
            return []                  # best-effort: degrade to [], never raise
        return self._findings_from_citations(task, budget)