Source Connector Framework

The source connector framework is a modular pull-based ingestion layer that fetches transcripts directly from external conversation platforms and delivers them to the CII intake pipeline. It allows new transcript sources to be added without modifying any existing API routes, domain models, or intake logic.

1. Architecture Overview

flowchart LR
    subgraph external["External Platforms"]
        FE[Fellow\nmeeting notes]
        GO[Gong\ncall transcripts]
        FU[Future\nsource...]
    end

    subgraph infra["infrastructure/connectors"]
        FC[FellowConnector\nfetch meetings\ntransform notes]
        GC[GongConnector\nfetch calls\nresolve speakers]
        PD[Provider Discovery\nprovider.py per package]
        REG[ConnectorRegistry\nsource_system → connector]
        SS[ConnectorSyncState\ncadence · baseline · watermark]
    end

    subgraph app["application/connectors"]
        CS[ConnectorService\nresolve window\nfetch → build artifact\n→ ingest]
    end

    subgraph api["api/v1/connectors"]
        R1["GET /v1/connectors"]
        R2["POST /v1/connectors/{system}/pull"]
        R3["POST /v1/connectors/{system}/fetch/{id}"]
        R4["GET/PUT /v1/connectors/{system}/sync"]
    end

    subgraph jobs["Celery"]
        J1[sync_due_connectors_task]
    end

    subgraph cii["CII Pipeline"]
        IS[IntakeService.ingest\nSegment · Tag · Extract\nTrend path]
        DB[(PostgreSQL)]
    end

    FE --> FC
    GO --> GC
    FU -.->|implement SourceConnector| REG
    FC --> REG
    GC --> REG
    PD --> REG
    SS --> CS
    REG --> CS
    CS --> IS
    IS --> DB
    R2 --> CS
    R3 --> CS
    R4 --> SS
    R1 --> REG
    J1 --> CS

Every connector returns ConnectedTranscript objects — a thin wrapper over the JSON transcript that the JSONSegmentationStrategy already understands. No custom segmentation logic is needed.

2. Core Abstractions

2.1 ConnectedTranscript

@dataclass(frozen=True)
class ConnectedTranscript:
    source_ref: str          # "fellow:mtg_001" | "gong:call_abc"
    source_system: str       # "fellow" | "gong"
    source_type: SourceType  # Always SourceType.JSON for connector output
    raw_content: str         # {"segments": [{"speaker": "...", "text": "..."}]}
    customer_scope: str | None = None
    external_created_at: datetime | None = None

raw_content is always the JSON format consumed by JSONSegmentationStrategy:

{
  "segments": [
    {
      "speaker": "Alice Smith",
      "text": "Let me walk through the Q1 roadmap.",
      "start": "120.5",
      "end": "128.3"
    }
  ]
}

2.2 SourceConnector Protocol

@runtime_checkable
class SourceConnector(Protocol):
    source_system: str

    async def fetch_since(
        self,
        since: datetime,
        *,
        until: datetime | None = None,
        customer_scope: str | None = None,
    ) -> list[ConnectedTranscript]: ...

    async def fetch_by_id(
        self, external_id: str, *, customer_scope: str | None = None
    ) -> ConnectedTranscript: ...

    async def close(self) -> None: ...

Any class that satisfies this structural interface is a valid connector. No base class inheritance required.

2.3 ConnectorRegistry

registry = ConnectorRegistry()
registry = build_connector_registry(settings)

connector = registry.get("fellow")
systems = registry.list_systems()

2.4 Connector Sync State

Each source system can have persisted sync state:

ConnectorSyncState(
    source_system="gong",
    enabled=True,
    cadence=ConnectorSyncCadence.DAILY,
    legal_basis=LegalBasis.LEGITIMATE_INTEREST,
    initial_since=datetime(2026, 6, 1, tzinfo=timezone.utc),
    last_successful_until=datetime(2026, 6, 9, 18, 0, tzinfo=timezone.utc),
)

This state drives default and scheduled pulls:

effective_since = last_successful_until when present, else initial_since
until = now unless an explicit upper bound is supplied
duplicates are still skipped by source_ref
watermark advances only after a clean pull with no non-duplicate ingest errors

2.5 ConnectorProvider Discovery

Each connector package exports a provider object from provider.py.

provider = ConnectorProvider(
    source_system="fellow",
    required_config_keys=("api_token",),
    default_config={"base_url": "https://api.fellow.ai/v2"},
    build_connector=_build_fellow_connector,
)

The application scans raw_to_knowledge.infrastructure.connectors.*.provider modules at runtime. If a package exposes a valid provider and the required config is present, the connector is instantiated automatically. This means a new connector can be added without editing route wiring or registry-construction logic.

2.6 Scheduled Sync

Celery beat runs a lightweight polling task on a fixed interval (CONNECTOR_SYNC_POLL_INTERVAL_MINUTES). That task:

Loads all enabled ConnectorSyncState rows
Checks each source's own cadence (hourly, daily, weekly)
Pulls only the due sources
Advances watermark on clean success or records last_error on failure

3. Fellow Connector

Source

Fellow is a meeting management platform. Its API provides meeting records and structured notes (text entries, action items, questions).

API base: https://api.fellow.ai/v2
Auth: Authorization: Bearer {api_token}
Setting: FELLOW_API_TOKEN

Transformation

Each Fellow meeting becomes one ConnectedTranscript. The meeting's note items are converted to segments:

Fellow field	Raw to Knowledge field
`note.author.name`	`segment.speaker`
`note.content` (HTML-stripped)	`segment.text`
`meeting.started_at`	`segment.start`
`meeting.ended_at`	`segment.end`
`"fellow:{meeting.id}"`	`source_ref`

Meetings with no non-empty notes are skipped.

Pagination

GET /v2/meetings?from=<ISO8601>&per_page=50&page=N — pages are fetched until the total count is reached.

Sequence

sequenceDiagram
    participant CS as ConnectorService
    participant FC as FellowConnector
    participant API as Fellow API v2
    participant IS as IntakeService

    CS->>FC: fetch_since(since, until, customer_scope)
    FC->>API: GET /v2/meetings?from=...
    API-->>FC: { meetings: [...], total: N }
    FC->>FC: _meeting_to_transcript() × N
    FC-->>CS: list[ConnectedTranscript]
    loop each transcript
        CS->>IS: ingest(SourceArtifact)
        IS-->>CS: IntakeResult
    end
    CS-->>caller: ConnectorPullResult

4. Gong Connector

Source

Gong is a conversation intelligence platform with AI-generated call transcripts. Each call transcript is structured as monologue blocks — one block per continuous stretch of a single speaker.

API base: https://api.gong.io/v2
Auth: Authorization: Basic {base64(access_key:access_secret)}
Settings: GONG_ACCESS_KEY, GONG_ACCESS_SECRET

Transformation

Each Gong call becomes one ConnectedTranscript. Each monologue block becomes one segment; all sentences in the block are joined with a space.

Gong field	Raw to Knowledge field
Resolved `speakerId` → user name	`segment.speaker`
Block sentences joined	`segment.text`
`sentences[0].start`	`segment.start`
`sentences[-1].end`	`segment.end`
`"gong:{callId}"`	`source_ref`

Speaker IDs are resolved to "{firstName} {lastName}" via GET /v2/users before transcript processing. If the user list call fails, the raw speaker ID is used as a fallback.

Calls with no transcript data are skipped.

Pagination

GET /v2/calls uses cursor-based pagination (records.cursor key).
POST /v2/calls/transcript batches up to 20 call IDs per request.
GET /v2/users uses cursor-based pagination.

Sequence

sequenceDiagram
    participant CS as ConnectorService
    participant GC as GongConnector
    participant API as Gong API v2
    participant IS as IntakeService

    CS->>GC: fetch_since(since, until, customer_scope)
    GC->>API: GET /v2/calls?fromDateTime=...
    API-->>GC: { records: { calls: [...], cursor } }
    GC->>API: GET /v2/users
    API-->>GC: { users: [...] }
    GC->>API: POST /v2/calls/transcript { filter: { callIds: [...] } }
    API-->>GC: { callsTranscripts: [...] }
    GC->>GC: _call_to_transcript() × N
    GC-->>CS: list[ConnectedTranscript]
    loop each transcript
        CS->>IS: ingest(SourceArtifact)
        IS-->>CS: IntakeResult
    end
    CS-->>caller: ConnectorPullResult

5. API Reference

GET /v1/connectors

Returns the source-system labels of all configured connectors.

{ "source_systems": ["fellow", "gong"] }

An empty list is returned if no connector credentials are configured.

POST /v1/connectors/{source_system}/pull

Pulls transcripts for either an explicit window or the source's persisted default window and ingests them.

Request:

{
  "since": "2026-06-01T00:00:00Z",
  "until": "2026-06-10T00:00:00Z",
  "legal_basis": "legitimate_interest",
  "customer_scope": "acme-corp"
}

Or, when sync state is configured, omit since and legal_basis to use the persisted default window:

{
  "customer_scope": "acme-corp"
}

Response:

{
  "source_system": "fellow",
  "since": "2026-06-01T00:00:00Z",
  "until": "2026-06-10T00:00:00Z",
  "fetched_count": 12,
  "ingested_count": 10,
  "skipped_count": 2,
  "watermark_advanced": true,
  "artifact_ids": ["art_001", "art_002", ...],
  "errors": []
}

Status	Meaning
`200`	Pull complete (partial errors reported in `errors` field)
`404`	Source system not registered / connector not configured
`422`	Invalid sync configuration or `legal_basis = NOT_ASSESSED`
`502`	Upstream API error (Fellow or Gong unreachable)

GET /v1/connectors/{source_system}/sync

Returns persisted sync cadence and watermark state for a configured source.

PUT /v1/connectors/{source_system}/sync

Creates or updates sync cadence and default pull configuration for a source.

Request:

{
  "enabled": true,
  "cadence": "daily",
  "legal_basis": "legitimate_interest",
  "customer_scope": "acme-corp",
  "initial_since": "2026-06-01T00:00:00Z"
}

POST /v1/connectors/{source_system}/fetch/{external_id}

Fetches and ingests a single transcript by its external ID.

Request:

{
  "legal_basis": "legitimate_interest",
  "customer_scope": "acme-corp"
}

Response:

{
  "artifact_id": "art_xyz",
  "source_system": "fellow",
  "external_id": "mtg_abc123"
}

Status	Meaning
`200`	Ingested successfully
`404`	Connector not registered, or item not found in the source system
`409`	Transcript already ingested (duplicate `source_ref`)
`502`	Upstream API error

6. Adding a New Connector — Complete Developer Guide

Adding a connector for a new source (Zoom, Teams, Fireflies, Chorus, Clari, etc.) follows an identical pattern every time. Nothing inside IntakeService, the domain layer, or the existing API routes changes.

6.1 File Layout

Create the connector package and tests:

# New files
src/raw_to_knowledge/infrastructure/connectors/{system}/
├── __init__.py          ← export the connector class
├── client.py            ← raw async HTTP client (all network I/O here)
├── connector.py         ← SourceConnector implementation (transform raw payloads)
└── provider.py          ← ConnectorProvider export for runtime discovery

tests/unit/connectors/
└── test_{system}_connector.py   ← unit tests (HTTP client mocked)

Replace {system} with the lowercase source-system label (e.g. zoom, teams, fireflies).

6.2 Step 1 — Write the HTTP Client

client.py does only one thing: make HTTP calls and return raw dicts. No Raw to Knowledge types here.

# src/raw_to_knowledge/infrastructure/connectors/{system}/client.py
from __future__ import annotations
from datetime import datetime
from typing import Any
import httpx
import structlog

log = structlog.get_logger()
_DEFAULT_BASE_URL = "https://api.{system}.example.com/v1"
_DEFAULT_TIMEOUT = 15.0


class {System}APIError(Exception):
    def __init__(self, status_code: int, detail: str) -> None:
        self.status_code = status_code
        super().__init__(f"{System} API error {status_code}: {detail}")


class {System}Client:
    def __init__(
        self,
        api_token: str,
        base_url: str = _DEFAULT_BASE_URL,
        timeout: float = _DEFAULT_TIMEOUT,
    ) -> None:
        self._api_token = api_token
        self._base_url = base_url.rstrip("/")
        self._timeout = timeout

    @property
    def _headers(self) -> dict[str, str]:
        # Adjust auth scheme for the target API
        return {
            "Authorization": f"Bearer {self._api_token}",
            "Accept": "application/json",
        }

    async def list_items(
        self,
        since: datetime,
        until: datetime | None = None,
    ) -> list[dict[str, Any]]:
        """Fetch all items created on or after ``since``.  Paginate as needed."""
        params: dict[str, Any] = {"from": since.strftime("%Y-%m-%dT%H:%M:%SZ")}
        if until:
            params["to"] = until.strftime("%Y-%m-%dT%H:%M:%SZ")

        all_items: list[dict[str, Any]] = []
        async with httpx.AsyncClient(headers=self._headers, timeout=self._timeout) as client:
            while True:
                resp = await self._get(client, "/items", params=params)
                items = resp.get("items") or []
                all_items.extend(items)
                # Replace with the API's actual pagination signal
                if not resp.get("next_page"):
                    break
                params["page"] = resp["next_page"]

        log.info("{system}_list_items_complete", count=len(all_items))
        return all_items

    async def get_item(self, item_id: str) -> dict[str, Any]:
        """Fetch a single item by its external ID."""
        async with httpx.AsyncClient(headers=self._headers, timeout=self._timeout) as client:
            return await self._get(client, f"/items/{item_id}")

    # ------------------------------------------------------------------
    async def _get(
        self,
        client: httpx.AsyncClient,
        path: str,
        params: dict | None = None,
    ) -> dict[str, Any]:
        url = f"{self._base_url}{path}"
        try:
            response = await client.get(url, params=params)
        except httpx.ConnectError as exc:
            raise {System}APIError(0, f"Connection failed: {exc}") from exc
        except httpx.TimeoutException as exc:
            raise {System}APIError(0, f"Request timed out: {exc}") from exc
        if response.status_code == 404:
            raise {System}APIError(404, f"Not found: {path}")
        if not response.is_success:
            raise {System}APIError(response.status_code, response.text[:500])
        return response.json()

Rules for the client:

Raise {System}APIError for every non-2xx response and every network failure.
Never import anything from raw_to_knowledge.domain or raw_to_knowledge.application.
The 404 case must set status_code=404 on the error — the connector uses this to distinguish "not found" from other failures.

6.3 Step 2 — Write the Connector

connector.py transforms the raw API responses into ConnectedTranscript objects. No HTTP calls here — those all delegate to the client.

# src/raw_to_knowledge/infrastructure/connectors/{system}/connector.py
from __future__ import annotations
from datetime import datetime, timezone
from typing import Any
import structlog

from raw_to_knowledge.domain.enums.intake import SourceType
from raw_to_knowledge.infrastructure.connectors.base import ConnectedTranscript
from raw_to_knowledge.infrastructure.connectors.{system}.client import {System}APIError, {System}Client

log = structlog.get_logger()


class {System}Connector:
    """Pull-based connector for the {System} platform."""

    source_system = "{system}"   # ← must be unique across all connectors

    def __init__(
        self,
        api_token: str,
        base_url: str = "https://api.{system}.example.com/v1",
        timeout: float = 15.0,
    ) -> None:
        self._client = {System}Client(api_token=api_token, base_url=base_url, timeout=timeout)

    async def fetch_since(
        self,
        since: datetime,
        *,
        customer_scope: str | None = None,
    ) -> list[ConnectedTranscript]:
        since_utc = since if since.tzinfo else since.replace(tzinfo=timezone.utc)
        try:
            items = await self._client.list_items(since=since_utc)
        except {System}APIError as exc:
            from raw_to_knowledge.domain.exceptions import ConnectorFetchError
            raise ConnectorFetchError("{system}", str(exc)) from exc

        transcripts = [
            t for item in items
            if (t := self._item_to_transcript(item, customer_scope=customer_scope)) is not None
        ]
        log.info("{system}_fetch_complete", since=since_utc.isoformat(),
                 total=len(items), produced=len(transcripts))
        return transcripts

    async def fetch_by_id(
        self,
        external_id: str,
        *,
        customer_scope: str | None = None,
    ) -> ConnectedTranscript:
        try:
            item = await self._client.get_item(external_id)
        except {System}APIError as exc:
            from raw_to_knowledge.domain.exceptions import ConnectorFetchError, ConnectorItemNotFoundError
            if exc.status_code == 404:
                raise ConnectorItemNotFoundError("{system}", external_id) from exc
            raise ConnectorFetchError("{system}", str(exc)) from exc

        transcript = self._item_to_transcript(item, customer_scope=customer_scope)
        if transcript is None:
            from raw_to_knowledge.domain.exceptions import ConnectorFetchError
            raise ConnectorFetchError("{system}", f"Item {external_id!r} has no usable content.")
        return transcript

    async def close(self) -> None:
        pass   # httpx clients are created per-request; nothing to release

    # ------------------------------------------------------------------
    def _item_to_transcript(
        self,
        item: dict[str, Any],
        *,
        customer_scope: str | None,
    ) -> ConnectedTranscript | None:
        item_id = item.get("id", "")

        # Build segments list — one dict per utterance/entry
        segments = []
        for entry in item.get("entries") or []:
            text = (entry.get("text") or "").strip()
            if not text:
                continue
            seg: dict[str, str] = {
                "speaker": (entry.get("author") or {}).get("name") or "Participant",
                "text": text,
            }
            if start := entry.get("start"):
                seg["start"] = str(start)
            if end := entry.get("end"):
                seg["end"] = str(end)
            segments.append(seg)

        if not segments:
            return None   # item has no usable content; will be skipped

        # Parse external creation timestamp if available
        external_created_at: datetime | None = None
        if created := item.get("created_at"):
            try:
                external_created_at = datetime.fromisoformat(str(created).replace("Z", "+00:00"))
            except (ValueError, TypeError):
                pass

        return ConnectedTranscript(
            source_ref=f"{{system}}:{item_id}",   # MUST be unique and stable
            source_system=self.source_system,
            source_type=SourceType.JSON,           # always JSON
            raw_content=ConnectedTranscript.build_json_content(segments),
            customer_scope=customer_scope,
            external_created_at=external_created_at,
        )

Critical invariants:

Invariant	Why
`source_ref = "{system}:{external_id}"`	The intake idempotency check uses this as the deduplication key. Changing the format breaks re-runs.
`source_type = SourceType.JSON`	The `ConnectorService` always builds a `SourceArtifact` with `SourceType.JSON`. `JSONSegmentationStrategy` handles it.
`raw_content = ConnectedTranscript.build_json_content(segments)`	Produces `{"segments": [...]}` — the exact format `JSONSegmentationStrategy` expects.
Items with no usable content return `None`	`ConnectorService._ingest_many` silently skips `None` results; do not raise here.
`fetch_by_id` raises `ConnectorItemNotFoundError` on 404	The API returns `404 Not Found` to the caller when the item is genuinely absent.
`fetch_by_id` raises `ConnectorFetchError` on non-404 API errors	The API returns `502 Bad Gateway` to the caller.

6.4 Step 3 — Create the Package Init

# src/raw_to_knowledge/infrastructure/connectors/{system}/__init__.py
"""{System} source connector."""
from raw_to_knowledge.infrastructure.connectors.{system}.connector import {System}Connector
__all__ = ["{System}Connector"]

6.5 Step 4 — Create the Provider

Add provider.py to the connector package:

# src/raw_to_knowledge/infrastructure/connectors/{system}/provider.py
from raw_to_knowledge.infrastructure.connectors.provider import ConnectorProvider
from raw_to_knowledge.infrastructure.connectors.{system}.connector import {System}Connector


def _build_{system}_connector(config: dict[str, str]) -> {System}Connector:
    return {System}Connector(
        api_token=config["api_token"],
        base_url=config["base_url"],
    )


provider = ConnectorProvider(
    source_system="{system}",
    required_config_keys=("api_token",),
    optional_config_keys=("base_url",),
    default_config={"base_url": "https://api.{system}.example.com/v1"},
    build_connector=_build_{system}_connector,
)

The provider is the discovery boundary. If the package exports this object, the application can find and build the connector automatically.

6.6 Step 5 — Configure the Connector

Provide credentials through the generic connector configuration map:

{
  "{system}": {
    "api_token": "secret-token",
    "base_url": "https://api.{system}.example.com/v1"
  }
}

Set that JSON into CONNECTOR_CONFIGS_JSON.

The connector is now live. With valid config set:

GET /v1/connectors returns "{system}" in the list
POST /v1/connectors/{system}/pull is operational
POST /v1/connectors/{system}/fetch/{id} is operational

Existing FELLOW_* and GONG_* variables remain supported as backward-compatible aliases.

6.7 Step 6 — Write Unit Tests

Mock the HTTP client at the method level so tests never touch the network. Follow the pattern from tests/unit/connectors/test_fellow_connector.py.

# tests/unit/connectors/test_{system}_connector.py
import json
import pytest
from datetime import datetime, timezone
from unittest.mock import AsyncMock

from raw_to_knowledge.domain.enums.intake import SourceType
from raw_to_knowledge.domain.exceptions import ConnectorFetchError, ConnectorItemNotFoundError
from raw_to_knowledge.infrastructure.connectors.{system}.client import {System}APIError
from raw_to_knowledge.infrastructure.connectors.{system}.connector import {System}Connector

SINCE = datetime(2026, 1, 1, tzinfo=timezone.utc)

# ---- sample API responses ----
ITEM_WITH_CONTENT = {
    "id": "item_001",
    "created_at": "2026-01-15T10:00:00Z",
    "entries": [
        {"text": "How does the API work?", "author": {"name": "Alice"}, "start": 0.0, "end": 5.0},
        {"text": "It follows REST conventions.", "author": {"name": "Bob"}, "start": 6.0, "end": 10.0},
    ],
}

ITEM_EMPTY = {"id": "item_002", "entries": []}


def _make_connector(items=None, single_item=None):
    connector = {System}Connector(api_token="tok_test")
    connector._client.list_items = AsyncMock(return_value=items or [])
    connector._client.get_item = AsyncMock(return_value=single_item or ITEM_WITH_CONTENT)
    return connector


# ---- fetch_since ----

@pytest.mark.asyncio
async def test_fetch_since_empty_returns_empty():
    assert await _make_connector([]).fetch_since(SINCE) == []

@pytest.mark.asyncio
async def test_fetch_since_source_ref_format():
    result = await _make_connector([ITEM_WITH_CONTENT]).fetch_since(SINCE)
    assert result[0].source_ref == "{system}:item_001"

@pytest.mark.asyncio
async def test_fetch_since_source_type_json():
    result = await _make_connector([ITEM_WITH_CONTENT]).fetch_since(SINCE)
    assert result[0].source_type == SourceType.JSON

@pytest.mark.asyncio
async def test_fetch_since_skips_empty_items():
    result = await _make_connector([ITEM_EMPTY]).fetch_since(SINCE)
    assert result == []

@pytest.mark.asyncio
async def test_fetch_since_segments_have_speaker_and_text():
    result = await _make_connector([ITEM_WITH_CONTENT]).fetch_since(SINCE)
    segs = json.loads(result[0].raw_content)["segments"]
    assert segs[0]["speaker"] == "Alice"
    assert "API" in segs[0]["text"]

@pytest.mark.asyncio
async def test_fetch_since_api_error_raises_connector_fetch_error():
    connector = {System}Connector(api_token="bad")
    connector._client.list_items = AsyncMock(side_effect={System}APIError(401, "Unauthorized"))
    with pytest.raises(ConnectorFetchError, match="{system}"):
        await connector.fetch_since(SINCE)

# ---- fetch_by_id ----

@pytest.mark.asyncio
async def test_fetch_by_id_returns_transcript():
    result = await _make_connector(single_item=ITEM_WITH_CONTENT).fetch_by_id("item_001")
    assert result.source_ref == "{system}:item_001"

@pytest.mark.asyncio
async def test_fetch_by_id_not_found_raises_item_not_found():
    connector = {System}Connector(api_token="tok")
    connector._client.get_item = AsyncMock(side_effect={System}APIError(404, "Not found"))
    with pytest.raises(ConnectorItemNotFoundError):
        await connector.fetch_by_id("missing")

@pytest.mark.asyncio
async def test_fetch_by_id_empty_content_raises_fetch_error():
    result = await _make_connector(single_item=ITEM_EMPTY).fetch_by_id("item_002")
    # Should raise ConnectorFetchError because no usable content
    connector = {System}Connector(api_token="tok")
    connector._client.get_item = AsyncMock(return_value=ITEM_EMPTY)
    with pytest.raises(ConnectorFetchError):
        await connector.fetch_by_id("item_002")

Minimum test coverage expected:

Test	What it verifies
`fetch_since` with empty list	Returns `[]` without error
`fetch_since` source_ref format	`"{system}:{id}"`
`fetch_since` source_type	Always `SourceType.JSON`
`fetch_since` skips empty items	Items with no content are excluded
`fetch_since` segment speaker + text	Transformation is correct
`fetch_since` API error → `ConnectorFetchError`	Error propagation
`fetch_by_id` success	Returns correct `source_ref`
`fetch_by_id` 404 → `ConnectorItemNotFoundError`	404 distinction
`fetch_by_id` empty content → `ConnectorFetchError`	No-content case
`raw_content` parses as valid JSON with `segments` key	Wire format correct
`customer_scope` propagated	Scope label flows through

6.8 Checklist

Copy this checklist into your pull request description:

Infrastructure
[ ] src/raw_to_knowledge/infrastructure/connectors/{system}/__init__.py created
[ ] src/raw_to_knowledge/infrastructure/connectors/{system}/client.py created
[ ] src/raw_to_knowledge/infrastructure/connectors/{system}/connector.py created
[ ] client.py raises {System}APIError on every non-2xx and network failure
[ ] client.py sets status_code=404 on 404 responses
[ ] client.py imports nothing from raw_to_knowledge.domain or raw_to_knowledge.application
[ ] connector.py raises ConnectorFetchError wrapping {System}APIError
[ ] connector.py raises ConnectorItemNotFoundError on 404 from fetch_by_id
[ ] source_system class attribute is a unique lowercase string
[ ] source_ref format is "{system}:{external_id}" (stable and unique)
[ ] source_type is always SourceType.JSON
[ ] raw_content uses ConnectedTranscript.build_json_content(segments)
[ ] Items with no usable segments return None (not raise) from _item_to_transcript

Configuration
[ ] Connector documented for CONNECTOR_CONFIGS_JSON usage
[ ] Any backward-compatible alias settings are documented only if intentionally added

Registration
[ ] `provider.py` exports a valid `ConnectorProvider`
[ ] Connector only registered when all required config values are non-empty
[ ] No route-wiring or registry-builder code changes required

Tests
[ ] tests/unit/connectors/test_{system}_connector.py created
[ ] HTTP client mocked at method level (no real network calls)
[ ] All minimum tests from §6.7 present
[ ] pytest --no-cov tests/unit/connectors/ passes with 0 failures

Documentation
[ ] docs/architecture/connectors.md §7 configuration table updated
[ ] docs/milestones/m1-intake.md deliverables table updated

7. Configuration Reference

Setting	Default	Description
`CONNECTOR_CONFIGS_JSON`	`""`	Generic JSON object keyed by source system. Used for discoverable connector configuration without adding new typed settings fields.
`FELLOW_API_TOKEN`	`""`	Fellow PAT or OAuth 2.0 bearer token; empty = connector not registered
`FELLOW_BASE_URL`	`https://api.fellow.ai/v2`	Override for testing or self-hosted instances
`GONG_ACCESS_KEY`	`""`	Gong API access key; both key+secret required to register
`GONG_ACCESS_SECRET`	`""`	Gong API access secret
`GONG_BASE_URL`	`https://api.gong.io/v2`	Override for testing

FELLOW_* and GONG_* settings are compatibility aliases. New connector packages should prefer CONNECTOR_CONFIGS_JSON.