feat: declare governed external provider state

AGENTS.md

@@ -0,0 +1,16 @@
+# GovOPlaN Addresses Codex Guide
+
+## Scope
+
+This repository owns reusable postal and electronic address records, address books, normalization, and governed address references for consuming modules.
+
+## Documentation Contract
+
+- Treat documentation as part of every behavior change. Update this module's manifest-driven `DocumentationTopic` contributions for affected user and administrator behavior.
+- Keep feature content here; `govoplan-docs` projects it without importing Addresses internals.
+- Maintain a static user/admin baseline and run `/mnt/DATA/git/govoplan/tools/checks/check-manifest-shapes.py` after behavior or manifest changes.
+
+## Boundaries
+
+- Keep identity and organization ownership in their respective modules.
+- Expose optional integrations through Core capabilities and typed references.

src/govoplan_addresses/backend/manifest.py

@@ -27,7 +27,18 @@ from govoplan_core.core.modules import (
     PermissionDefinition,
     RoleTemplate,
 )
+from govoplan_core.core.provider_governance import (
+    ExternalProviderDeclaration,
+    ExternalProviderStateProviderRegistration,
+    ProviderBehaviorDeclaration,
+    ProviderObjectDeclaration,
+    declared_module_architecture,
+)
 from govoplan_core.db.base import Base
+from govoplan_addresses.backend.provider_state import (
+    CARDDAV_PROVIDER_ID,
+    carddav_provider_states,
+)
 
 
 _addresses_table_retirement_provider = drop_table_retirement_provider(
@@ -152,6 +163,59 @@ def _addresses_router(_context: ModuleContext):
     return router
 
 
+CARDDAV_PROVIDER = ExternalProviderDeclaration(
+    id=CARDDAV_PROVIDER_ID,
+    module_id="addresses",
+    label="CardDAV address-book synchronization",
+    maturity="synchronize",
+    operations=("discover", "read", "write", "delete", "synchronize", "preview"),
+    objects=(
+        ProviderObjectDeclaration(
+            object_type="address_book",
+            field_groups=("identity", "display", "sync_state"),
+            authority_modes=("external_authoritative", "external_mirror", "governed_sync"),
+            default_authority_mode="external_mirror",
+        ),
+        ProviderObjectDeclaration(
+            object_type="contact",
+            field_groups=("identity", "name", "postal", "email", "phone", "source_metadata"),
+            authority_modes=("external_authoritative", "external_mirror", "governed_sync"),
+            default_authority_mode="governed_sync",
+        ),
+    ),
+    behavior=ProviderBehaviorDeclaration(
+        revision_tokens="CardDAV sync tokens, resource hrefs, and ETags are retained.",
+        concurrency="Conditional writes reject stale ETags and preserve explicit conflicts.",
+        freshness="Last attempt, last success, source status, and sync token are recorded.",
+        health="Transport failures, diagnostics, and unresolved conflicts are projected separately.",
+        max_read_items=5000,
+        idempotency="Stable source, href, UID, and ETag facts prevent duplicate contact effects.",
+        retry="Only a new governed sync attempt retries failed transport operations.",
+        timeout_seconds=30,
+        conflicts="Local and remote values remain in an explicit conflict record until resolved.",
+        outcome_unknown="Timed-out writes require a subsequent CardDAV read before correction or retry.",
+        outcome_unknown_supported=True,
+        evidence="Sync diagnostics, tombstones, conflicts, source revisions, and contact provenance are retained.",
+        audit_event_types=(
+            "addresses.sync.started",
+            "addresses.sync.finished",
+            "addresses.sync.conflict_recorded",
+        ),
+        correction="A resolved conflict or later synchronized revision corrects state without rewriting prior evidence.",
+        rollback="Remote writes are not assumed to be transactionally reversible.",
+        compensation="A reconciled update or tombstone can compensate after the remote outcome is known.",
+        reconciliation="Read by resource href and compare ETag, UID, and local revision before applying changes.",
+        outage="Existing local contacts remain available with stale or unknown freshness.",
+        classifications=("personal", "confidential"),
+        purposes=("address-book synchronization", "governed recipient resolution"),
+        retention="Address-book and audit retention policies apply independently.",
+        secret_handling="Only credential references and sanitized authentication metadata are persisted in sync state.",
+    ),
+    capability_names=(CAPABILITY_ADDRESSES_LOOKUP, CAPABILITY_ADDRESSES_CONTACT_WRITER),
+    documentation_topic_ids=("addresses.boundary",),
+)
+
+
 manifest = ModuleManifest(
     id="addresses",
     name="Addresses",
@@ -234,6 +298,33 @@ manifest = ModuleManifest(
             order=30,
         ),
     ),
+    external_providers=(CARDDAV_PROVIDER,),
+    external_provider_state_providers=(
+        ExternalProviderStateProviderRegistration(
+            module_id="addresses",
+            provider_id=CARDDAV_PROVIDER_ID,
+            provider=carddav_provider_states,
+        ),
+    ),
+    architecture=declared_module_architecture(
+        layer="communication_participation",
+        kind="domain",
+        maturity="vertical_slice",
+        documentation_ref="docs/ADDRESS_MODULE_ARCHITECTURE.md",
+        test_ref="tests/test_addresses_service.py",
+        known_limits=("External address-book synchronization remains a bounded connector slice rather than a supported provider profile.",),
+        supported_authority_modes=(
+            "native_authoritative",
+            "external_authoritative",
+            "external_mirror",
+            "governed_sync",
+        ),
+        owned_concepts=("contact point", "address book", "contact consent", "recipient source"),
+        non_owned_concepts=("identity", "organization", "campaign recipient snapshot", "procedure party"),
+        target_tested_providers=(CARDDAV_PROVIDER_ID,),
+        security_docs=("docs/ADDRESS_MODULE_ARCHITECTURE.md",),
+        operations_docs=("README.md",),
+    ),
 )
 
 

src/govoplan_addresses/backend/provider_state.py

@@ -0,0 +1,164 @@
+from __future__ import annotations
+
+from collections import defaultdict
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from govoplan_addresses.backend.db.models import (
+    AddressSyncConflict,
+    AddressSyncDiagnostic,
+    AddressSyncSource,
+)
+from govoplan_core.core.provider_governance import (
+    ExternalProviderRuntimeState,
+    ExternalProviderStateContext,
+)
+
+
+CARDDAV_PROVIDER_ID = "addresses.carddav_sync"
+_CURRENT_WINDOW = timedelta(hours=24)
+
+
+def carddav_provider_states(
+    context: ExternalProviderStateContext,
+) -> tuple[ExternalProviderRuntimeState, ...]:
+    if not isinstance(context.session, Session):
+        raise RuntimeError("Addresses provider state requires a database session.")
+    statement = select(AddressSyncSource).where(
+        AddressSyncSource.connector_type == "carddav"
+    )
+    if context.tenant_id is not None:
+        statement = statement.where(AddressSyncSource.tenant_id == context.tenant_id)
+    sources = tuple(
+        context.session.scalars(
+            statement.order_by(AddressSyncSource.tenant_id, AddressSyncSource.id).limit(
+                context.max_items + 1
+            )
+        )
+    )
+    if not sources:
+        return ()
+
+    source_ids = tuple(item.id for item in sources)
+    conflict_counts = _grouped_counts(
+        context.session,
+        AddressSyncConflict.sync_source_id,
+        AddressSyncConflict.status == "open",
+        source_ids,
+    )
+    error_counts = _grouped_counts(
+        context.session,
+        AddressSyncDiagnostic.sync_source_id,
+        AddressSyncDiagnostic.severity == "error",
+        source_ids,
+    )
+    observed_at = datetime.now(UTC)
+    return tuple(
+        _source_state(
+            source,
+            observed_at=observed_at,
+            conflict_count=conflict_counts.get(source.id, 0),
+            error_count=error_counts.get(source.id, 0),
+        )
+        for source in sources
+    )
+
+
+def _grouped_counts(
+    session: Session,
+    source_column: object,
+    predicate: object,
+    source_ids: tuple[str, ...],
+) -> dict[str, int]:
+    counts: dict[str, int] = defaultdict(int)
+    rows = session.execute(
+        select(source_column, func.count()).where(
+            source_column.in_(source_ids), predicate
+        ).group_by(source_column)
+    )
+    for source_id, count in rows:
+        counts[str(source_id)] = int(count)
+    return counts
+
+
+def _source_state(
+    source: AddressSyncSource,
+    *,
+    observed_at: datetime,
+    conflict_count: int,
+    error_count: int,
+) -> ExternalProviderRuntimeState:
+    active = bool(source.enabled)
+    status = str(source.status or "idle")
+    health = (
+        "inactive"
+        if not active
+        else "error"
+        if status == "failed" or bool(source.last_error)
+        else "warning"
+        if status in {"conflict", "running"} or conflict_count or error_count
+        else "healthy"
+        if status == "succeeded"
+        else "unknown"
+    )
+    freshness = _freshness(source, observed_at=observed_at)
+    conflict = "pending" if conflict_count or status == "conflict" else "clear"
+    recovery = (
+        "not_applicable"
+        if not active
+        else "attention"
+        if health in {"error", "warning"} or conflict == "pending"
+        else "ready"
+    )
+    return ExternalProviderRuntimeState(
+        provider_id=CARDDAV_PROVIDER_ID,
+        binding_ref=f"addresses:sync-source:{source.id}",
+        authority_mode=(
+            "external_mirror" if source.read_only else "governed_sync"
+        ),
+        observed_at=observed_at,
+        configured=True,
+        active=active,
+        health=health,
+        freshness=freshness,
+        conflict=conflict,
+        recovery=recovery,
+        last_success_at=_aware(source.last_success_at),
+        detail=(
+            "CardDAV source is disabled."
+            if not active
+            else "CardDAV source requires reconciliation."
+            if conflict == "pending"
+            else "CardDAV source health has not been observed yet."
+            if health == "unknown"
+            else "CardDAV source state is available."
+        ),
+        metrics={
+            "open_conflicts": conflict_count,
+            "error_diagnostics": error_count,
+            "read_only": bool(source.read_only),
+            "status": status,
+        },
+    )
+
+
+def _freshness(source: AddressSyncSource, *, observed_at: datetime) -> str:
+    if not source.enabled:
+        return "not_applicable"
+    last_success = _aware(source.last_success_at)
+    if last_success is None:
+        return "unknown"
+    return "current" if observed_at - last_success <= _CURRENT_WINDOW else "stale"
+
+
+def _aware(value: datetime | None) -> datetime | None:
+    if value is None:
+        return None
+    if value.tzinfo is None:
+        return value.replace(tzinfo=UTC)
+    return value.astimezone(UTC)
+
+
+__all__ = ["CARDDAV_PROVIDER_ID", "carddav_provider_states"]

tests/test_provider_state.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from govoplan_addresses.backend.db.models import (
+    AddressBook,
+    AddressSyncConflict,
+    AddressSyncDiagnostic,
+    AddressSyncSource,
+)
+from govoplan_addresses.backend.manifest import manifest
+from govoplan_addresses.backend.provider_state import (
+    CARDDAV_PROVIDER_ID,
+    carddav_provider_states,
+)
+from govoplan_core.core.provider_governance import ExternalProviderStateContext
+from govoplan_core.db.base import Base
+
+
+class AddressesProviderStateTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
+        Base.metadata.create_all(
+            self.engine,
+            tables=(
+                AddressBook.__table__,
+                AddressSyncSource.__table__,
+                AddressSyncConflict.__table__,
+                AddressSyncDiagnostic.__table__,
+            ),
+        )
+        self.session = sessionmaker(bind=self.engine, expire_on_commit=False)()
+        book = AddressBook(
+            id="book-1",
+            tenant_id="tenant-1",
+            scope_type="tenant",
+            scope_id="tenant-1",
+            name="Remote",
+        )
+        self.source = AddressSyncSource(
+            id="source-1",
+            tenant_id="tenant-1",
+            address_book_id=book.id,
+            connector_type="carddav",
+            display_name="CardDAV",
+            external_address_book_ref="https://dav.example.test/addressbook/",
+            sync_direction="two_way",
+            read_only=False,
+            enabled=True,
+            status="succeeded",
+            last_success_at=datetime.now(UTC),
+        )
+        self.session.add_all((book, self.source))
+        self.session.commit()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_state_is_tenant_bounded_secret_free_and_reports_conflict(self) -> None:
+        healthy = carddav_provider_states(
+            ExternalProviderStateContext(session=self.session, tenant_id="tenant-1")
+        )[0]
+
+        self.assertEqual("healthy", healthy.health)
+        self.assertEqual("current", healthy.freshness)
+        self.assertEqual("governed_sync", healthy.authority_mode)
+        self.assertNotIn("dav.example.test", str(healthy.to_dict()))
+        self.assertEqual(
+            (),
+            carddav_provider_states(
+                ExternalProviderStateContext(
+                    session=self.session,
+                    tenant_id="tenant-2",
+                )
+            ),
+        )
+
+        self.session.add(
+            AddressSyncConflict(
+                tenant_id="tenant-1",
+                sync_source_id=self.source.id,
+                address_book_id="book-1",
+                field_path="email",
+                status="open",
+            )
+        )
+        self.session.flush()
+
+        conflicted = carddav_provider_states(
+            ExternalProviderStateContext(session=self.session, tenant_id="tenant-1")
+        )[0]
+        self.assertEqual("pending", conflicted.conflict)
+        self.assertEqual("attention", conflicted.recovery)
+
+    def test_manifest_registers_carddav_declaration_and_state(self) -> None:
+        self.assertEqual(CARDDAV_PROVIDER_ID, manifest.external_providers[0].id)
+        self.assertEqual(
+            CARDDAV_PROVIDER_ID,
+            manifest.external_provider_state_providers[0].provider_id,
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()