diff --git a/docs/MODULE_ARCHITECTURE.md b/docs/MODULE_ARCHITECTURE.md
index 33ffbae..802258d 100644
--- a/docs/MODULE_ARCHITECTURE.md
+++ b/docs/MODULE_ARCHITECTURE.md
@@ -1491,6 +1491,22 @@ The first implementation is a platform access gate. It does not replace
 database backups, process supervision, migration checks, or external load
 balancer maintenance pages.
 
+## Connector Runtime Contract
+
+Core defines provider-neutral connector preview and diagnostic primitives in
+`govoplan_core.core.connector_runtime`. The contract keeps optional modules
+decoupled: Connectors owns transport, endpoint discovery, retries, and protocol
+health; the consuming domain module owns mappings, validation, reconciliation,
+and mutations of its records.
+
+Every dry run is bounded and identifies the source revision, source fingerprint,
+immutable input hash, effects, and redacted diagnostics. Its summary must match
+the returned effect list exactly. An apply token is usable only when the preview
+is complete, current, conflict-free, and contains no error diagnostic. Endpoint
+URLs never contain credentials; only credential-envelope references cross the
+contract. Provider-specific details belong in sanitized provenance rather than
+in a shared domain schema.
+
 ## Build And Verification
 
 Backend verification from core:
diff --git a/src/govoplan_core/core/connector_runtime.py b/src/govoplan_core/core/connector_runtime.py
new file mode 100644
index 0000000..d06a334
--- /dev/null
+++ b/src/govoplan_core/core/connector_runtime.py
@@ -0,0 +1,226 @@
+from __future__ import annotations
+
+from collections import Counter
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Literal, Protocol, runtime_checkable
+from urllib.parse import urlsplit
+
+
+CONNECTOR_RUNTIME_CONTRACT_VERSION = "1.0"
+
+ConnectorDiagnosticSeverity = Literal["info", "warning", "error"]
+ConnectorDiagnosticStage = Literal[
+    "configuration",
+    "authentication",
+    "discovery",
+    "read",
+    "mapping",
+    "planning",
+    "apply",
+    "reconciliation",
+]
+ConnectorEffectKind = Literal[
+    "create",
+    "update",
+    "delete",
+    "conflict",
+    "unchanged",
+    "ignored",
+]
+ConnectorOutcomeState = Literal[
+    "preview",
+    "accepted",
+    "rejected",
+    "outcome_unknown",
+]
+
+
+class ConnectorContractError(ValueError):
+    pass
+
+
+@dataclass(frozen=True, slots=True)
+class ConnectorEndpoint:
+    """Sanitized endpoint identity. Credentials never belong in this value."""
+
+    url: str
+    credential_ref: str | None = None
+    tls_mode: Literal["required", "start_tls", "system", "disabled"] = "required"
+    options: Mapping[str, object] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        normalized = self.url.strip()
+        parsed = urlsplit(normalized)
+        if not parsed.scheme or not parsed.hostname:
+            raise ConnectorContractError("Connector endpoints require an absolute URL.")
+        if parsed.username is not None or parsed.password is not None:
+            raise ConnectorContractError(
+                "Connector endpoint URLs must not contain credentials."
+            )
+        object.__setattr__(self, "url", normalized)
+        if self.credential_ref is not None:
+            credential_ref = self.credential_ref.strip()
+            if not credential_ref:
+                raise ConnectorContractError("Credential references cannot be blank.")
+            object.__setattr__(self, "credential_ref", credential_ref)
+
+
+@dataclass(frozen=True, slots=True)
+class ConnectorDiagnostic:
+    severity: ConnectorDiagnosticSeverity
+    code: str
+    message: str
+    stage: ConnectorDiagnosticStage
+    retryable: bool = False
+    source_ref: str | None = None
+    object_ref: str | None = None
+    details: Mapping[str, object] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        if not self.code.strip() or len(self.code) > 120:
+            raise ConnectorContractError(
+                "Connector diagnostic codes must contain 1 to 120 characters."
+            )
+        if not self.message.strip():
+            raise ConnectorContractError("Connector diagnostic messages cannot be blank.")
+
+
+@dataclass(frozen=True, slots=True)
+class ConnectorEffectPreview:
+    effect: ConnectorEffectKind
+    source_object_ref: str
+    target_object_ref: str | None = None
+    changed_fields: tuple[str, ...] = ()
+    sample: Mapping[str, object] = field(default_factory=dict)
+    reason_code: str | None = None
+    outcome: ConnectorOutcomeState = "preview"
+    revision: str | None = None
+
+    def __post_init__(self) -> None:
+        if not self.source_object_ref.strip():
+            raise ConnectorContractError("Preview effects require a source object reference.")
+        if self.outcome != "preview":
+            raise ConnectorContractError("Dry-run effects must retain the preview outcome.")
+
+
+@dataclass(frozen=True, slots=True)
+class ConnectorEffectSummary:
+    creates: int = 0
+    updates: int = 0
+    deletes: int = 0
+    conflicts: int = 0
+    unchanged: int = 0
+    ignored: int = 0
+
+    @property
+    def total(self) -> int:
+        return (
+            self.creates
+            + self.updates
+            + self.deletes
+            + self.conflicts
+            + self.unchanged
+            + self.ignored
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class ConnectorDryRunRequest:
+    tenant_id: str
+    source_ref: str
+    force_full: bool = False
+    max_items: int = 1_000
+    expected_source_revision: str | None = None
+    context: Mapping[str, object] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        if not self.tenant_id.strip() or not self.source_ref.strip():
+            raise ConnectorContractError("Dry runs require tenant and source references.")
+        if not 1 <= self.max_items <= 10_000:
+            raise ConnectorContractError("Dry-run max_items must be between 1 and 10000.")
+
+
+@dataclass(frozen=True, slots=True)
+class ConnectorDryRunResult:
+    contract_version: str
+    source_ref: str
+    source_revision: str
+    source_fingerprint: str
+    input_hash: str
+    generated_at: datetime
+    summary: ConnectorEffectSummary
+    effects: tuple[ConnectorEffectPreview, ...] = ()
+    diagnostics: tuple[ConnectorDiagnostic, ...] = ()
+    truncated: bool = False
+    stale: bool = False
+    apply_token: str | None = None
+    provenance: Mapping[str, object] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        if self.contract_version != CONNECTOR_RUNTIME_CONTRACT_VERSION:
+            raise ConnectorContractError(
+                f"Unsupported connector contract version: {self.contract_version!r}."
+            )
+        for name in ("source_ref", "source_revision", "source_fingerprint", "input_hash"):
+            if not str(getattr(self, name)).strip():
+                raise ConnectorContractError(f"Dry-run {name} cannot be blank.")
+        if self.summary.total != len(self.effects):
+            raise ConnectorContractError(
+                "Dry-run summary counts must match the returned effect list."
+            )
+
+    @property
+    def can_apply(self) -> bool:
+        return (
+            self.apply_token is not None
+            and not self.truncated
+            and not self.stale
+            and self.summary.conflicts == 0
+            and not any(item.severity == "error" for item in self.diagnostics)
+        )
+
+
+def summarize_connector_effects(
+    effects: tuple[ConnectorEffectPreview, ...],
+) -> ConnectorEffectSummary:
+    counts = Counter(item.effect for item in effects)
+    return ConnectorEffectSummary(
+        creates=counts["create"],
+        updates=counts["update"],
+        deletes=counts["delete"],
+        conflicts=counts["conflict"],
+        unchanged=counts["unchanged"],
+        ignored=counts["ignored"],
+    )
+
+
+@runtime_checkable
+class ConnectorDryRunProvider(Protocol):
+    def preview(
+        self,
+        session: object,
+        principal: object,
+        *,
+        request: ConnectorDryRunRequest,
+    ) -> ConnectorDryRunResult:
+        ...
+
+
+__all__ = [
+    "CONNECTOR_RUNTIME_CONTRACT_VERSION",
+    "ConnectorContractError",
+    "ConnectorDiagnostic",
+    "ConnectorDiagnosticSeverity",
+    "ConnectorDiagnosticStage",
+    "ConnectorDryRunProvider",
+    "ConnectorDryRunRequest",
+    "ConnectorDryRunResult",
+    "ConnectorEffectKind",
+    "ConnectorEffectPreview",
+    "ConnectorEffectSummary",
+    "ConnectorEndpoint",
+    "ConnectorOutcomeState",
+    "summarize_connector_effects",
+]
diff --git a/tests/test_connector_runtime_contract.py b/tests/test_connector_runtime_contract.py
new file mode 100644
index 0000000..4f19a3c
--- /dev/null
+++ b/tests/test_connector_runtime_contract.py
@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+import unittest
+from datetime import UTC, datetime
+
+from govoplan_core.core.connector_runtime import (
+    CONNECTOR_RUNTIME_CONTRACT_VERSION,
+    ConnectorContractError,
+    ConnectorDryRunResult,
+    ConnectorEffectPreview,
+    ConnectorEndpoint,
+    summarize_connector_effects,
+)
+
+
+class ConnectorRuntimeContractTests(unittest.TestCase):
+    def test_endpoint_rejects_embedded_credentials(self) -> None:
+        with self.assertRaisesRegex(ConnectorContractError, "must not contain credentials"):
+            ConnectorEndpoint(url="ldaps://user:secret@directory.example.test")
+
+    def test_dry_run_is_bounded_consistent_and_apply_gated(self) -> None:
+        effects = (
+            ConnectorEffectPreview(
+                effect="create",
+                source_object_ref="ldap:uid=one",
+                sample={"display_name": "Example Person"},
+            ),
+            ConnectorEffectPreview(
+                effect="unchanged",
+                source_object_ref="ldap:uid=two",
+                target_object_ref="contact-2",
+            ),
+        )
+        result = ConnectorDryRunResult(
+            contract_version=CONNECTOR_RUNTIME_CONTRACT_VERSION,
+            source_ref="source-1",
+            source_revision="revision-1",
+            source_fingerprint="a" * 64,
+            input_hash="b" * 64,
+            generated_at=datetime(2026, 8, 2, tzinfo=UTC),
+            summary=summarize_connector_effects(effects),
+            effects=effects,
+            apply_token="plan-1",
+        )
+
+        self.assertEqual(1, result.summary.creates)
+        self.assertEqual(1, result.summary.unchanged)
+        self.assertTrue(result.can_apply)
+
+    def test_summary_must_describe_exact_returned_effects(self) -> None:
+        with self.assertRaisesRegex(ConnectorContractError, "summary counts"):
+            ConnectorDryRunResult(
+                contract_version=CONNECTOR_RUNTIME_CONTRACT_VERSION,
+                source_ref="source-1",
+                source_revision="revision-1",
+                source_fingerprint="a" * 64,
+                input_hash="b" * 64,
+                generated_at=datetime(2026, 8, 2, tzinfo=UTC),
+                summary=summarize_connector_effects(()),
+                effects=(
+                    ConnectorEffectPreview(
+                        effect="create",
+                        source_object_ref="source:one",
+                    ),
+                ),
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()