feat: declare governed external provider state

2026-08-01 17:48:25 +02:00
parent 52fe33568c
commit 7be93785a2
6 changed files with 578 additions and 12 deletions
@@ -20,6 +20,15 @@ from govoplan_core.core.modules import (
    PermissionDefinition,
    RoleTemplate,
 )
+from govoplan_core.core.provider_governance import (
+    ExternalProviderDeclaration,
+    ExternalProviderStateProviderRegistration,
+    ModuleArchitectureDeclaration,
+    ModuleArchitectureDocumentation,
+    ModuleMaturityEvidence,
+    ProviderBehaviorDeclaration,
+    ProviderObjectDeclaration,
+)
 from govoplan_core.core.tabular_sources import (
    CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER,
    CAPABILITY_CONNECTORS_TABULAR_SOURCES,
@@ -48,6 +57,12 @@ from govoplan_connectors.backend.datasource_origins import (
    ConnectorDatasourceOriginProvider,
 )
 from govoplan_connectors.backend.feeds import ConnectorFeedProvider
+from govoplan_connectors.backend.provider_state import (
+    SANCTIONS_PROVIDER_ID,
+    TABULAR_PROVIDER_ID,
+    sanctions_provider_states,
+    tabular_provider_states,
+)


 MODULE_ID = "connectors"
@@ -57,6 +72,142 @@ DATASOURCE_ORIGIN_INTERFACE_VERSION = "0.1.0"
 SANCTIONS_SNAPSHOT_INTERFACE_VERSION = "1.0.0"
 FEED_INTERFACE_VERSION = "0.1.0"

+ARCHITECTURE = ModuleArchitectureDeclaration(
+    layer="data_reporting_integration",
+    kind="integration",
+    maturity="vertical_slice",
+    evidence=(
+        ModuleMaturityEvidence(
+            kind="test",
+            reference="tests/test_tabular_sources.py",
+            summary="Exercises tenant-safe immutable tabular snapshots and bounded reads.",
+        ),
+        ModuleMaturityEvidence(
+            kind="test",
+            reference="tests/test_sanctions_sources.py",
+            summary="Exercises source acquisition health, checksums, retries, and immutable evidence.",
+        ),
+        ModuleMaturityEvidence(
+            kind="documentation",
+            reference="docs/CONNECTOR_SOURCE_LIFECYCLE.md",
+            summary="Defines source lifecycle, authority, evidence, and outage boundaries.",
+        ),
+    ),
+    known_limits=(
+        "The executable generic datasource origin is an immutable tabular snapshot; database and arbitrary REST profiles remain future providers.",
+        "Feed publication renders a governed document but does not yet push it to an external publishing endpoint.",
+    ),
+    supported_authority_modes=(
+        "external_authoritative",
+        "external_mirror",
+        "linked_reference",
+    ),
+    owned_concepts=(
+        "external transport profiles",
+        "protocol interaction",
+        "immutable connector snapshots",
+        "connector acquisition health",
+    ),
+    non_owned_concepts=(
+        "datasource catalogue identity and lifecycle",
+        "domain records and business semantics",
+        "data transformations",
+        "screening dispositions",
+    ),
+    target_tested_providers=(
+        TABULAR_PROVIDER_ID,
+        SANCTIONS_PROVIDER_ID,
+    ),
+    documentation=ModuleArchitectureDocumentation(
+        migration=("src/govoplan_connectors/backend/migrations/versions",),
+        upgrade=("docs/CONNECTOR_SOURCE_LIFECYCLE.md",),
+        recovery=("docs/CONNECTOR_SOURCE_LIFECYCLE.md",),
+        security=("docs/CONNECTOR_SOURCE_LIFECYCLE.md",),
+        operations=("docs/CONNECTOR_SOURCE_LIFECYCLE.md",),
+    ),
+)
+
+EXTERNAL_PROVIDERS = (
+    ExternalProviderDeclaration(
+        id=TABULAR_PROVIDER_ID,
+        module_id=MODULE_ID,
+        label="Immutable tabular snapshot provider",
+        maturity="read",
+        operations=("discover", "search", "read", "preview", "dry_run"),
+        objects=(
+            ProviderObjectDeclaration(
+                object_type="tabular_source_snapshot",
+                field_groups=("identity", "schema", "rows", "source_provenance"),
+                authority_modes=("external_authoritative", "external_mirror"),
+                default_authority_mode="external_mirror",
+            ),
+        ),
+        behavior=ProviderBehaviorDeclaration(
+            revision_tokens="Source fingerprints and immutable snapshot ids are retained.",
+            concurrency="Reads may require the expected fingerprint; snapshots never mutate in place.",
+            freshness="Snapshot acquisition time and source timestamp are exposed.",
+            health="Import validation and source-read failures are explicit.",
+            max_read_items=1000,
+            evidence="Rows, schema, fingerprint, source metadata, and acquisition provenance remain linked.",
+            correction="Import a replacement snapshot; retain the prior snapshot as evidence.",
+            reconciliation="Compare source and snapshot fingerprints before selecting a new current state.",
+            outage="Existing snapshots remain available and visibly stale; no live-source claim is made.",
+            classifications=("internal", "confidential", "restricted"),
+            purposes=("governed import", "dataflow input", "evidence reconstruction"),
+            retention="Datasources or the consuming domain supplies retention and hold policy.",
+            secret_handling="Generic snapshots contain no connector credential; transport credentials stay in credential envelopes.",
+        ),
+        capability_names=(
+            CAPABILITY_CONNECTORS_TABULAR_SOURCES,
+            CAPABILITY_DATASOURCE_ORIGINS,
+        ),
+        interface_names=(
+            "connectors.tabular_sources",
+            "connectors.datasource_origins",
+        ),
+        documentation_topic_ids=(
+            "connectors.authority-and-effects",
+            "connectors.tabular-sources",
+        ),
+    ),
+    ExternalProviderDeclaration(
+        id=SANCTIONS_PROVIDER_ID,
+        module_id=MODULE_ID,
+        label="Sanctions source snapshot provider",
+        maturity="read",
+        operations=("discover", "search", "read", "preview"),
+        objects=(
+            ProviderObjectDeclaration(
+                object_type="sanctions_source_snapshot",
+                field_groups=("source_identity", "raw_evidence", "entries", "acquisition_health"),
+                authority_modes=("external_authoritative", "external_mirror"),
+                default_authority_mode="external_mirror",
+            ),
+        ),
+        behavior=ProviderBehaviorDeclaration(
+            revision_tokens="Provider source version, ETag, Last-Modified, and SHA-256 digest are retained when available.",
+            concurrency="Refreshes use conditional source requests and create immutable snapshots.",
+            freshness="Latest successful acquisition, source timestamp, and stale health are reported.",
+            health="Transport, parsing, source-change, and malformed-source states are explicit.",
+            max_read_items=5000,
+            evidence="Raw source bytes, checksum, acquisition run, parser result, and normalized entry count are linked.",
+            correction="A corrected source creates a new immutable snapshot and acquisition run.",
+            reconciliation="Compare source version and digest, then preserve both prior and corrected evidence.",
+            outage="The latest accepted snapshot stays usable with stale/unavailable source health.",
+            classifications=("public", "internal"),
+            purposes=("sanctions source acquisition", "compliance screening evidence"),
+            retention="Risk and Records policies determine accepted snapshot retention and legal holds.",
+            secret_handling="Public sources require no subject data or source credential; configured proxy secrets remain external to snapshots.",
+        ),
+        capability_names=(CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS,),
+        interface_names=("connectors.sanctions_snapshots",),
+        documentation_topic_ids=(
+            "connectors.authority-and-effects",
+            "connectors.sanctions-snapshots",
+        ),
+    ),
+)
+

 def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
    module_id, resource, action = scope.split(":", 2)
@@ -157,17 +308,12 @@ def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
        ),
        "connector_sanctions_snapshots": (
            session.query(ConnectorSanctionsSnapshot)
-            .filter(
-                ConnectorSanctionsSnapshot.tenant_id == tenant_id
-            )
+            .filter(ConnectorSanctionsSnapshot.tenant_id == tenant_id)
            .count()
        ),
        "connector_sanctions_runs": (
            session.query(ConnectorSanctionsAcquisitionRun)
-            .filter(
-                ConnectorSanctionsAcquisitionRun.tenant_id
-                == tenant_id
-            )
+            .filter(ConnectorSanctionsAcquisitionRun.tenant_id == tenant_id)
            .count()
        ),
    }
@@ -220,12 +366,24 @@ manifest = ModuleManifest(
        CAPABILITY_CONNECTORS_TABULAR_SOURCES: _provider,
        CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER: _provider,
        CAPABILITY_DATASOURCE_ORIGINS: _datasource_origin_provider,
-        CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS: (
-            _sanctions_snapshot_provider
-        ),
+        CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS: (_sanctions_snapshot_provider),
        CAPABILITY_CONNECTORS_FEEDS: _feed_provider,
    },
    tenant_summary_providers=(_tenant_summary,),
+    architecture=ARCHITECTURE,
+    external_providers=EXTERNAL_PROVIDERS,
+    external_provider_state_providers=(
+        ExternalProviderStateProviderRegistration(
+            module_id=MODULE_ID,
+            provider_id=TABULAR_PROVIDER_ID,
+            provider=tabular_provider_states,
+        ),
+        ExternalProviderStateProviderRegistration(
+            module_id=MODULE_ID,
+            provider_id=SANCTIONS_PROVIDER_ID,
+            provider=sanctions_provider_states,
+        ),
+    ),
    migration_spec=MigrationSpec(
        module_id=MODULE_ID,
        metadata=Base.metadata,
@@ -251,6 +409,21 @@ manifest = ModuleManifest(
        ),
    ),
    documentation=(
+        DocumentationTopic(
+            id="connectors.authority-and-effects",
+            title="Connector authority and effect behavior",
+            summary="Connector direction, technical maturity, and configured source authority are separate and must remain visible.",
+            body=(
+                "A connector can consume, publish, or work bidirectionally and can mature from discovery through replacement. "
+                "Each binding separately states whether GovOPlaN is authoritative, follows an external authority, keeps a mirror, synchronizes under conflict rules, adds a governance overlay, or retains only a link. "
+                "Writable providers must explain revisions, limits, idempotency, outcome-unknown handling, evidence, reconciliation, correction, outage behavior, and secret requirements."
+            ),
+            layer="available",
+            documentation_types=("admin", "user"),
+            audience=("operator", "module_admin", "power_user", "product_owner"),
+            related_modules=("datasources", "dataflow", "ops", "policy", "audit"),
+            order=39,
+        ),
        DocumentationTopic(
            id="connectors.tabular-sources",
            title="Governed tabular sources",
@@ -0,0 +1,207 @@
+from __future__ import annotations
+
+from collections import defaultdict
+from datetime import UTC, datetime
+from hashlib import sha256
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from govoplan_connectors.backend.db.models import (
+    ConnectorSanctionsAcquisitionRun,
+    ConnectorSanctionsSnapshot,
+    ConnectorTabularSource,
+)
+from govoplan_core.core.provider_governance import (
+    ExternalProviderRuntimeState,
+    ExternalProviderStateContext,
+)
+
+
+TABULAR_PROVIDER_ID = "connectors.tabular_snapshot"
+SANCTIONS_PROVIDER_ID = "connectors.sanctions_snapshot"
+
+
+def tabular_provider_states(
+    context: ExternalProviderStateContext,
+) -> tuple[ExternalProviderRuntimeState, ...]:
+    session = _session(context)
+    statement = select(ConnectorTabularSource).where(
+        ConnectorTabularSource.deleted_at.is_(None)
+    )
+    if context.tenant_id is not None:
+        statement = statement.where(
+            ConnectorTabularSource.tenant_id == context.tenant_id
+        )
+    sources = tuple(
+        session.scalars(
+            statement.order_by(
+                ConnectorTabularSource.tenant_id,
+                ConnectorTabularSource.id,
+            ).limit(context.max_items + 1)
+        )
+    )
+    observed_at = datetime.now(UTC)
+    return tuple(_tabular_state(item, observed_at=observed_at) for item in sources)
+
+
+def sanctions_provider_states(
+    context: ExternalProviderStateContext,
+) -> tuple[ExternalProviderRuntimeState, ...]:
+    session = _session(context)
+    statement = select(ConnectorSanctionsAcquisitionRun)
+    if context.tenant_id is not None:
+        statement = statement.where(
+            ConnectorSanctionsAcquisitionRun.tenant_id == context.tenant_id
+        )
+    runs = tuple(
+        session.scalars(
+            statement.order_by(
+                ConnectorSanctionsAcquisitionRun.tenant_id,
+                ConnectorSanctionsAcquisitionRun.provider_id,
+                ConnectorSanctionsAcquisitionRun.source_id,
+                ConnectorSanctionsAcquisitionRun.started_at.desc(),
+            ).limit(max(context.max_items * 10, context.max_items + 1))
+        )
+    )
+    latest_by_binding: dict[tuple[str, str, str], ConnectorSanctionsAcquisitionRun] = {}
+    for run in runs:
+        key = (run.tenant_id, run.provider_id, run.source_id)
+        latest_by_binding.setdefault(key, run)
+        if len(latest_by_binding) >= context.max_items + 1:
+            break
+
+    snapshot_counts = _snapshot_counts(
+        session,
+        binding_keys=tuple(latest_by_binding),
+    )
+    observed_at = datetime.now(UTC)
+    return tuple(
+        _sanctions_state(
+            run,
+            observed_at=observed_at,
+            snapshot_count=snapshot_counts.get(key, 0),
+        )
+        for key, run in latest_by_binding.items()
+    )
+
+
+def _session(context: ExternalProviderStateContext) -> Session:
+    if not isinstance(context.session, Session):
+        raise RuntimeError("Connectors provider state requires a database session.")
+    return context.session
+
+
+def _tabular_state(
+    source: ConnectorTabularSource,
+    *,
+    observed_at: datetime,
+) -> ExternalProviderRuntimeState:
+    active = source.status == "active"
+    return ExternalProviderRuntimeState(
+        provider_id=TABULAR_PROVIDER_ID,
+        binding_ref=f"connectors:tabular-source:{source.id}",
+        authority_mode="external_mirror",
+        observed_at=observed_at,
+        configured=True,
+        active=active,
+        health="healthy" if active else "inactive",
+        freshness="not_applicable",
+        conflict="not_applicable",
+        recovery="ready" if active else "not_applicable",
+        last_success_at=_aware(source.updated_at or source.created_at),
+        detail=(
+            "Immutable tabular snapshot is available."
+            if active
+            else "Immutable tabular snapshot is inactive."
+        ),
+        metrics={
+            "row_count": int(source.row_count),
+            "byte_count": int(source.byte_count),
+            "schema_version": int(source.schema_version),
+        },
+    )
+
+
+def _snapshot_counts(
+    session: Session,
+    *,
+    binding_keys: tuple[tuple[str, str, str], ...],
+) -> dict[tuple[str, str, str], int]:
+    if not binding_keys:
+        return {}
+    tenant_ids = {item[0] for item in binding_keys}
+    rows = session.execute(
+        select(
+            ConnectorSanctionsSnapshot.tenant_id,
+            ConnectorSanctionsSnapshot.provider_id,
+            ConnectorSanctionsSnapshot.source_id,
+            func.count(ConnectorSanctionsSnapshot.id),
+        )
+        .where(ConnectorSanctionsSnapshot.tenant_id.in_(tenant_ids))
+        .group_by(
+            ConnectorSanctionsSnapshot.tenant_id,
+            ConnectorSanctionsSnapshot.provider_id,
+            ConnectorSanctionsSnapshot.source_id,
+        )
+    )
+    return {
+        (str(tenant_id), str(provider_id), str(source_id)): int(count)
+        for tenant_id, provider_id, source_id, count in rows
+        if (str(tenant_id), str(provider_id), str(source_id)) in binding_keys
+    }
+
+
+def _sanctions_state(
+    run: ConnectorSanctionsAcquisitionRun,
+    *,
+    observed_at: datetime,
+    snapshot_count: int,
+) -> ExternalProviderRuntimeState:
+    status = str(run.status)
+    success = status in {"succeeded", "success", "not_modified"}
+    running = status in {"running", "pending", "retry"}
+    has_snapshot = bool(run.snapshot_id) or snapshot_count > 0
+    health = "healthy" if success else "warning" if running else "error"
+    binding_digest = sha256(
+        f"{run.tenant_id}\0{run.provider_id}\0{run.source_id}".encode("utf-8")
+    ).hexdigest()[:24]
+    return ExternalProviderRuntimeState(
+        provider_id=SANCTIONS_PROVIDER_ID,
+        binding_ref=f"connectors:sanctions-source:{binding_digest}",
+        authority_mode="external_mirror",
+        observed_at=observed_at,
+        configured=True,
+        active=True,
+        health=health,
+        freshness="unknown",
+        conflict="not_applicable",
+        recovery="ready" if success and has_snapshot else "attention",
+        last_success_at=_aware(run.finished_at) if success else None,
+        detail=(
+            "Latest sanctions acquisition completed."
+            if success
+            else "Sanctions acquisition is in progress."
+            if running
+            else "Latest sanctions acquisition failed; prior accepted snapshots remain separate evidence."
+        ),
+        metrics={
+            "latest_status": status,
+            "attempt_count": int(run.attempt_count),
+            "accepted_snapshots": int(snapshot_count),
+        },
+    )
+
+
+def _aware(value: datetime | None) -> datetime | None:
+    if value is None:
+        return None
+    return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)
+
+
+__all__ = [
+    "SANCTIONS_PROVIDER_ID",
+    "TABULAR_PROVIDER_ID",
+    "sanctions_provider_states",
+    "tabular_provider_states",
+]