feat: expose connector datasource origins

.gitignore

@@ -0,0 +1,276 @@
+__pycache__/
+*.py[cod]
+*.egg-info/
+.pytest_cache/
+.mypy_cache/
+.ruff_cache/
+.venv/
+build/
+dist/
+node_modules/
+webui/node_modules/
+webui/dist/
+*.tsbuildinfo
+.component-test-build/
+.module-test-build/
+.policy-test-build/
+.template-preview-test-build/
+.import-test-build/
+webui/.component-test-build/
+webui/.module-test-build/
+webui/.policy-test-build/
+webui/.template-preview-test-build/
+webui/.import-test-build/
+
+# GovOPlaN shared ignore rules from govoplan-core
+# ---> Node
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+lerna-debug.log*
+.pnpm-debug.log*
+# Diagnostic reports (https://nodejs.org/api/report.html)
+report.[0-9]*.[0-9]*.[0-9]*.[0-9]*.json
+# Runtime data
+pids
+*.pid
+*.seed
+*.pid.lock
+# Directory for instrumented libs generated by jscoverage/JSCover
+lib-cov
+# Coverage directory used by tools like istanbul
+coverage
+*.lcov
+# nyc test coverage
+.nyc_output
+# Grunt intermediate storage (https://gruntjs.com/creating-plugins#storing-task-files)
+.grunt
+# Bower dependency directory (https://bower.io/)
+bower_components
+# node-waf configuration
+.lock-wscript
+# Compiled binary addons (https://nodejs.org/api/addons.html)
+build/Release
+# Dependency directories
+jspm_packages/
+# Snowpack dependency directory (https://snowpack.dev/)
+web_modules/
+# TypeScript cache
+# Optional npm cache directory
+.npm
+# Optional eslint cache
+.eslintcache
+# Optional stylelint cache
+.stylelintcache
+# Microbundle cache
+.rpt2_cache/
+.rts2_cache_cjs/
+.rts2_cache_es/
+.rts2_cache_umd/
+# Optional REPL history
+.node_repl_history
+# Output of 'npm pack'
+*.tgz
+# Yarn Integrity file
+.yarn-integrity
+# dotenv environment variable files
+.env
+.env.development.local
+.env.test.local
+.env.production.local
+.env.local
+# parcel-bundler cache (https://parceljs.org/)
+.cache
+.parcel-cache
+# Next.js build output
+.next
+out
+# Nuxt.js build / generate output
+.nuxt
+dist
+# Gatsby files
+.cache/
+# Comment in the public line in if your project uses Gatsby and not Next.js
+# https://nextjs.org/blog/next-9-1#public-directory-support
+# public
+# vuepress build output
+.vuepress/dist
+# vuepress v2.x temp and cache directory
+.temp
+.cache
+# vitepress build output
+**/.vitepress/dist
+# vitepress cache directory
+**/.vitepress/cache
+# Docusaurus cache and generated files
+.docusaurus
+# Serverless directories
+.serverless/
+# FuseBox cache
+.fusebox/
+# DynamoDB Local files
+.dynamodb/
+# TernJS port file
+.tern-port
+# Stores VSCode versions used for testing VSCode extensions
+.vscode-test
+# yarn v2
+.yarn/cache
+.yarn/unplugged
+.yarn/build-state.yml
+.yarn/install-state.gz
+.pnp.*
+# Local WebUI test/build scratch directories
+# ---> Python
+# Byte-compiled / optimized / DLL files
+*$py.class
+# C extensions
+*.so
+# Distribution / packaging
+.Python
+develop-eggs/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+.installed.cfg
+*.egg
+MANIFEST
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+cover/
+# Translations
+*.mo
+*.pot
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+# Flask stuff:
+instance/
+.webassets-cache
+# Scrapy stuff:
+.scrapy
+# Sphinx documentation
+docs/_build/
+# PyBuilder
+.pybuilder/
+target/
+# Jupyter Notebook
+.ipynb_checkpoints
+# IPython
+profile_default/
+ipython_config.py
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+# UV
+#   Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#uv.lock
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/latest/usage/project/#working-with-version-control
+.pdm.toml
+.pdm-python
+.pdm-build/
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+# SageMath parsed files
+*.sage.py
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+# Spyder project settings
+.spyderproject
+.spyproject
+# Rope project settings
+.ropeproject
+# mkdocs documentation
+/site
+# mypy
+.dmypy.json
+dmypy.json
+# Pyre type checker
+.pyre/
+# pytype static type analyzer
+.pytype/
+# Cython debug symbols
+cython_debug/
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
+# Ruff stuff:
+# PyPI configuration file
+.pypirc
+# ---> VisualStudioCode
+.vscode/*
+!.vscode/settings.json
+!.vscode/tasks.json
+!.vscode/launch.json
+!.vscode/extensions.json
+!.vscode/*.code-snippets
+# Local History for Visual Studio Code
+.history/
+# Built Visual Studio Code Extensions
+*.vsix
+*.db
+# GovOPlaN local runtime state
+runtime/
+# GovOPlaN WebUI test output

README.md

@@ -0,0 +1,44 @@
+# govoplan-connectors
+
+<!-- govoplan-repository-type:start -->
+**Repository type:** connector (connector-hub).
+<!-- govoplan-repository-type:end -->
+
+`govoplan-connectors` owns integration catalogues and generic external system
+connection patterns for GovOPlaN.
+
+The module should make external systems discoverable, testable, and usable
+without taking ownership of their business semantics. Domain-specific modules
+remain responsible for case, file, workflow, payment, mail, identity, document,
+or reporting behavior.
+
+## Executable First Slice
+
+The first executable connector capability provides tenant-isolated tabular
+origins. Operators can import bounded JSON or CSV snapshots, inspect inferred
+schemas, and expose immutable source references and content fingerprints
+through `connectors.datasource_origins@0.1.0`.
+
+Connectors owns acquisition, connection profiles, credentials, discovery, and
+provider health. `govoplan-datasources` registers an origin as a governed live
+or cached datasource and owns staging, materializations, frozen states, and
+consumer access. Dataflow consumes that Datasources contract and never imports
+connector implementations or stores connector credentials.
+
+Database, REST/HTTP, directory, managed-file, and warehouse providers can
+implement the same origin contract without changing Datasources or Dataflow.
+
+Development:
+
+```bash
+/mnt/DATA/git/govoplan/.venv/bin/python -m pip install -e .
+/mnt/DATA/git/govoplan/.venv/bin/python -m unittest discover -s tests
+```
+
+See:
+
+- [Connector concept](docs/CONCEPT.md)
+- [Public-sector integration catalogue](docs/PUBLIC_SECTOR_INTEGRATION_CATALOGUE.md)
+- [Connector source lifecycle](docs/CONNECTOR_SOURCE_LIFECYCLE.md)
+- [OpenProject connector concept](docs/OPENPROJECT_CONNECTOR.md)
+- [OpenDesk integration map](docs/OPENDESK_INTEGRATION_MAP.md)

docs/CONCEPT.md

@@ -0,0 +1,170 @@
+# govoplan-connectors Concept
+
+## Purpose
+
+`govoplan-connectors` is the integration catalogue and connector coordination
+module. It helps GovOPlaN connect to existing public-sector and organizational
+systems without pretending to replace every specialist platform.
+
+The module owns connector metadata, connection profiles, health checks, test
+results, credential references, and generic integration events. Protocol-heavy
+or domain-heavy integrations may live in dedicated modules once their scope is
+clear.
+
+Detailed follow-up documents:
+
+- [Public-sector integration catalogue](PUBLIC_SECTOR_INTEGRATION_CATALOGUE.md)
+- [Connector source lifecycle](CONNECTOR_SOURCE_LIFECYCLE.md)
+- [OpenProject connector concept](OPENPROJECT_CONNECTOR.md)
+- [OpenDesk integration map](OPENDESK_INTEGRATION_MAP.md)
+
+## Ownership
+
+The module owns:
+
+- connector catalogue entries and capability metadata
+- connection profiles and endpoint configuration
+- credential references and test diagnostics
+- generic webhook/polling/job coordination metadata
+- connector health status and last-test evidence
+- operator-visible integration inventory
+- cross-module discovery of available external capabilities
+
+The module does not own:
+
+- governed datasource identity, staging, materializations, frozen states, or
+  consumer read semantics, owned by `govoplan-datasources`
+- file storage semantics, owned by files/DMS
+- identity provisioning semantics, owned by IDM/access
+- mail/calendar semantics, owned by mail/calendar
+- case/workflow/task/domain records
+- payment, ledger, XRechnung, XTA/OSCI, FIT-Connect, or XOE/V protocol
+  semantics once those are dedicated modules
+
+## Connector Categories
+
+Initial catalogue categories:
+
+- project management and task systems such as OpenProject
+- DMS/e-file/archive systems
+- file providers such as Nextcloud, Seafile, WebDAV, SMB/NFS, object storage
+- identity providers such as LDAP, Active Directory, OIDC, SAML, OpenDesk IDM
+- groupware such as Open-Xchange mail/calendar
+- ERP, finance, accounting, payment, and cash-register systems
+- public-sector protocols such as FIT-Connect, XTA/OSCI, XRechnung, XOE/V
+- reporting, BI, RSS/API publication, and open-data endpoints
+
+## Core Contracts
+
+The module should integrate through:
+
+- module manifest metadata, route factories, permissions, and migrations
+- a connector catalogue API for listing available connector types
+- a connection profile API with secret references, not plaintext secrets
+- capability declarations such as `connectors.catalog`,
+  `connectors.profileTester`, and `connectors.health`
+- events such as `connector.profile_created`, `connector.test_succeeded`,
+  `connector.test_failed`, and `connector.health_changed`
+- configuration-package fragments for required external systems
+
+Domain modules should ask whether a connector capability exists and request a
+profile/test result through core-mediated capabilities. They must not import
+connector implementation modules directly.
+
+Data-oriented consumers use a two-layer path: Connectors publishes a
+provider-specific datasource origin, then Datasources registers and governs it.
+Dataflow, Workflow, Reporting, and other consumers use Datasources rather than
+calling the connector origin directly.
+
+## Reference Journeys
+
+### OpenProject Connector First
+
+1. Operator registers an OpenProject connection profile.
+2. Connector tests API reachability and authentication.
+3. A future project-management decision can use the connector before a native
+   `govoplan-projects` module exists.
+4. Cases/tasks/workflow may link to external project/task references through
+   stable external-reference DTOs.
+
+### Public-Sector Integration Catalogue
+
+1. Operator records which external systems exist in an organization.
+2. GovOPlaN identifies common protocols and missing connectors.
+3. Configuration packages can declare required connector profiles.
+4. Health/status pages show whether required integrations are ready.
+
+### OpenDesk Profile
+
+1. Operator records OpenDesk component profiles for identity, mail, calendar,
+   files/documents, and OpenProject where present.
+2. Connectors shows which components are configured, tested, degraded, or
+   missing.
+3. Domain modules enable optional behavior by checking capabilities through
+   core, not by importing connector or OpenDesk-specific implementation code.
+
+## MVP Slice
+
+The first implementation should provide:
+
+- connector type registry
+- connection profile CRUD with secret references
+- connection test result records
+- WebUI catalogue and profile pages
+- configuration-package fragment support
+- generic external-reference DTOs
+- health summary provider
+
+## Permissions
+
+Candidate scopes:
+
+- `connectors:catalog:read`
+- `connectors:profile:read`
+- `connectors:profile:write`
+- `connectors:profile:test`
+- `connectors:secret:manage`
+- `connectors:admin`
+
+## Data Model Sketch
+
+Candidate tables:
+
+- `connector_types`
+- `connector_profiles`
+- `connector_profile_tests`
+- `connector_health_status`
+- `external_references`
+
+Plaintext credentials must never be stored in connector tables. Use secret
+references and the platform secret contract.
+
+## WebUI
+
+Initial route contributions:
+
+- `/connectors`
+- `/connectors/profiles/:profileId`
+
+The UI should show profile status, last test result, capability labels, required
+configuration-package dependencies, and external-reference search where a
+connector supports it.
+
+## Tests
+
+Minimum tests:
+
+- core starts with connectors installed and no domain modules present
+- profile validation rejects plaintext secret echoing
+- connection tests record success/failure diagnostics without leaking secrets
+- configuration package can require a connector profile
+- domain-module optional behavior can detect connector capabilities without
+  imports
+
+## Open Decisions
+
+- Whether protocol-specific connector modules depend on `govoplan-connectors`
+  or only share kernel contracts.
+- Which connector type should be first after OpenProject.
+- How much polling/webhook scheduling belongs here versus workflow/ops.
+- Whether external-reference indexing should move to search/dataflow later.

docs/CONNECTOR_SOURCE_LIFECYCLE.md

@@ -0,0 +1,162 @@
+# Connector Source Lifecycle
+
+GovOPlaN modules should treat external systems as sources with explicit
+lifecycle state. A connector profile can consume records from a source, publish
+records into a source, or do both. The lifecycle below keeps connectors
+predictable and avoids hidden module imports.
+
+## Source Directions
+
+- `consume`: GovOPlaN reads external records, normalizes them, and exposes them
+  to modules as external references, events, or staged imports.
+- `publish`: GovOPlaN creates or updates external records and stores the external
+  identifiers as immutable references.
+- `bidirectional`: GovOPlaN supports both directions with conflict detection and
+  reconciliation rules.
+
+## Source Data Lifecycle
+
+Connector profiles have operational states, while individual external records
+or source datasets move through a data lifecycle:
+
+1. `discovered`
+   A source, record, file, feed item, webhook event, or remote object is known
+   but not yet trusted for domain use.
+2. `connected`
+   GovOPlaN can authenticate and fetch or publish against the source profile.
+3. `imported`
+   Minimal source data has been staged with external id, version/ETag, source
+   timestamp, and provenance.
+4. `validated`
+   Shape, permissions, freshness, and required fields passed connector and
+   domain validation.
+5. `transformed`
+   A dataflow, workflow, or domain module normalized the staged payload into a
+   domain-specific form.
+6. `published`
+   GovOPlaN exposed or wrote an output through API, RSS, report, export, or a
+   downstream connector.
+7. `archived`
+   The source/output is no longer active but remains available under retention,
+   audit, and external-reference rules.
+8. `deprecated`
+   The source/output remains readable for history but must not be used for new
+   workflows.
+
+Every transition must preserve provenance, permissions context, freshness, and
+audit trace. Domain modules may add stricter states, but they should map back to
+this lifecycle when a connector publishes status.
+
+## Lifecycle States
+
+1. `draft`
+   Profile exists but is not used by runtime jobs.
+2. `configured`
+   Required endpoint and credential references are present.
+3. `tested`
+   A health/test run succeeded and recorded non-secret diagnostics.
+4. `active`
+   Runtime jobs may consume or publish data.
+5. `degraded`
+   The connector is active but health checks or recent jobs show failures.
+6. `paused`
+   Operators intentionally stop scheduled connector activity.
+7. `retiring`
+   The connector is being removed from active workflows while references remain
+   readable.
+8. `retired`
+   No new runtime activity is allowed. Historical references remain available.
+
+## State Transition Gates
+
+| Transition | Required Evidence | Blockers |
+| --- | --- | --- |
+| `draft` -> `configured` | endpoint fields are valid, credential references exist, owner/tenant scope is set | plaintext secret in profile payload, unsupported connector type |
+| `configured` -> `tested` | latest profile test succeeded and diagnostics were redacted | failed auth, unreachable endpoint, TLS/policy error |
+| `tested` -> `active` | operator enabled runtime use, required modules/capabilities are present, schedule/webhook is valid | missing module, missing permission, no idempotency strategy for publish jobs |
+| `active` -> `degraded` | health check or job telemetry reports failures | none; this is automatic diagnostic state |
+| `degraded` -> `active` | health/test succeeds or failed jobs are reconciled | unresolved conflict or repeated failure threshold |
+| any running state -> `paused` | operator pause request or maintenance preflight | active critical transaction that cannot be interrupted |
+| `paused` -> `active` | successful re-test when credentials/endpoints changed | failed profile test |
+| any state -> `retiring` | uninstall/disable plan accepted, schedulers/workers stopped | active domain references that require operator decision |
+| `retiring` -> `retired` | non-destructive retirement complete, references remain readable | destructive retirement requested without provider and backup |
+
+## Consume Flow
+
+1. Discover changes through polling, webhook, batch upload, or manual operator
+   action.
+2. Fetch only the minimal remote data required for the declared use case.
+3. Normalize into a connector-owned staging payload.
+4. Validate shape, required fields, and source trust level.
+5. Publish data-shaped inputs as versioned datasource origins.
+6. Let Datasources register live/cached origins or stage immutable snapshots.
+7. Let domain modules consume governed datasource references through
+   capabilities, not imports.
+8. Emit a core-mediated event such as `connector.record_discovered`.
+9. Store external references with source system, object type, object id, version
+   or ETag, and last-seen timestamp.
+
+## Publish Flow
+
+1. Domain module requests publish through a core-mediated connector capability.
+2. Connector validates profile state, permission, idempotency key, and payload
+   shape.
+3. Connector sends the remote request.
+4. Connector stores the remote id, version/ETag, and response diagnostics.
+5. Connector emits `connector.record_published` or `connector.publish_failed`.
+6. Domain module stores only the external-reference DTO and any domain result.
+
+## Reconciliation
+
+Every connector that writes to an external system needs a reconciliation story:
+
+- idempotency key for create/update jobs
+- remote object version, ETag, or last-modified value where available
+- conflict state when local and remote records diverge
+- retry policy for temporary failures
+- explicit operator action for destructive overwrite or deletion
+- audit trace from GovOPlaN record to external request and response summary
+
+## Capability Boundary
+
+Domain modules must not import connector implementation packages directly. They
+should ask core for capabilities such as:
+
+- `connectors.catalog`
+- `connectors.profileTester`
+- `connectors.health`
+- `connectors.externalReferences`
+- `connectors.datasourceOrigins`
+- `connectors.sourceConsumer`
+- `connectors.sourcePublisher`
+
+Connector payloads should be DTOs or protocol objects from kernel/core
+contracts. Protocol-specific clients stay inside the connector module that owns
+them.
+
+## Safety Rules
+
+- Secret values never leave the secret contract and are never stored in test
+  result payloads.
+- Runtime jobs must include profile id, connector type, direction, idempotency
+  key, and triggering principal/system actor.
+- Profile tests must redact tokens, passwords, cookies, authorization headers,
+  and remote personal data not needed for diagnostics.
+- Deactivation must stop schedulers/workers before profile removal.
+- Uninstall defaults to non-destructive retirement; domain data and external
+  references remain readable.
+- Destructive retirement requires a module-owned retirement provider and an
+  explicit operator choice.
+
+## Release Checklist
+
+Before shipping an executable connector type:
+
+- Add catalogue metadata and capability names.
+- Add profile schema validation that rejects plaintext secrets.
+- Add redaction tests for success and failure diagnostics.
+- Add unavailable-optional-module tests for every consuming domain module.
+- Add profile test and health status fixtures.
+- Add external-reference DTO tests.
+- Add lifecycle transition tests for pause, retry, retirement, and uninstall
+  guard behavior.

docs/GOVERNED_CONNECTOR_CONFIGURATION.md

@@ -0,0 +1,48 @@
+# Governed Connector Configuration
+
+GovOPlaN connectors should make integration behavior inspectable and testable.
+The target is not hardcoded glue hidden in module code, but governed connector
+definitions with schemas, mappings, test runs, simulation, versioning, and
+audit-visible execution.
+
+## Connector Definition
+
+A connector definition should describe:
+
+- provider type and protocol
+- endpoint and credential requirements
+- supported capabilities
+- input and output schemas
+- mapping and transformation versions
+- validation rules
+- dry-run and test operations
+- privacy and retention classification
+- expected events and audit records
+- operational limits and retry behavior
+
+Provider-specific code may still be required, but the configured integration
+logic should remain visible and reviewable.
+
+## Runtime Expectations
+
+Connectors should support:
+
+- discovery where possible
+- typed configuration through UI-managed controls
+- secret references instead of plaintext secrets
+- dry-run plans before writes
+- simulation with sample payloads
+- provenance for consumed and produced data
+- idempotent external writes where supported
+- quarantine/manual-review state for unsafe or ambiguous results
+
+Configuration packages may install connector definitions, but local overrides
+must be protected from accidental package updates.
+
+## Relationship To Datasources And Dataflow
+
+Recurring extraction and transformation should start as configuration across
+connectors, files, workflow, reporting, and templates. Create dedicated
+datasource or dataflow modules only when repeated source-catalog, lineage,
+mapping, scheduling, or publication contracts clearly outgrow connector
+ownership.

docs/OPENDESK_INTEGRATION_MAP.md

@@ -0,0 +1,71 @@
+# OpenDesk Integration Map
+
+OpenDesk is an integration profile across GovOPlaN modules, not a monolithic
+GovOPlaN module. The profile should let an operator see which OpenDesk
+components are connected, which GovOPlaN module owns each behavior, and which
+optional capabilities are available.
+
+The core boundary decision register is in
+`/mnt/DATA/git/govoplan-core/docs/MODULE_ARCHITECTURE.md`.
+
+## Component Routing
+
+| OpenDesk area | Example component | GovOPlaN owner | Integration behavior |
+| --- | --- | --- | --- |
+| Identity and directory | OpenDesk IDM, LDAP, AD, OIDC, SAML, SCIM | `govoplan-idm`, `govoplan-access` | integrate, synchronize selected accounts/groups, map principals and memberships |
+| Mail/groupware | Open-Xchange mail | `govoplan-mail` | integrate, link profiles, test mailbox/send/append, keep mail semantics in mail |
+| Calendar/groupware | Open-Xchange calendar, CalDAV/CardDAV | `govoplan-calendar` | integrate, free/busy lookup, selected event sync, resource calendars |
+| Files/documents | Nextcloud/WebDAV/files, office integrations | `govoplan-files`, later `govoplan-dms` | integrate, import/link files, keep document lifecycle in DMS |
+| Project management | OpenProject | `govoplan-connectors`, consumers in tasks/workflow/cases | connector-first, link/synchronize selected work packages |
+| Portal/collaboration | Portal/chat/video/office services where present | `govoplan-portal`, `govoplan-connectors`, `govoplan-dms`, `govoplan-workflow` | link/integrate only when a process needs it |
+| Inventory and diagnostics | endpoint catalogue, profile health, version checks | `govoplan-connectors` | catalogue, profile test, health summary, optional capability discovery |
+
+## Integration Behavior
+
+- `integrate`: call a stable API or protocol for the component.
+- `link`: store external references and open the external tool for
+  source-of-truth work.
+- `import`: bring selected files/records into GovOPlaN-owned storage or
+  evidence.
+- `synchronize`: keep selected records aligned through explicit source-of-truth
+  rules.
+- `replace selected workflow`: only when GovOPlaN owns tighter governance,
+  audit, retention, or configuration-package state than the OpenDesk component.
+
+## Shared Assumptions
+
+- Identity is the first dependency. Mail, calendar, files, and project
+  connectors should record which identity profile or tenant mapping they expect.
+- Connector profiles store references to secrets, never secret values.
+- Module consumers discover optional behavior through core capabilities and
+  module metadata.
+- The profile must work partially: an installation can have OpenProject without
+  Open-Xchange, or calendar without files.
+
+## Candidate Profile Shape
+
+```json
+{
+  "id": "opendesk-main",
+  "display_name": "OpenDesk",
+  "components": {
+    "identity": {"profile_id": "opendesk-idm", "owner": "govoplan-idm"},
+    "mail": {"profile_id": "ox-mail", "owner": "govoplan-mail"},
+    "calendar": {"profile_id": "ox-calendar", "owner": "govoplan-calendar"},
+    "files": {"profile_id": "nextcloud-main", "owner": "govoplan-files"},
+    "projects": {"profile_id": "openproject-main", "owner": "govoplan-connectors"}
+  }
+}
+```
+
+## Follow-Up Implementation Issues
+
+Existing high-priority module issues:
+
+- `govoplan-idm#1`: LDAP, Active Directory, OpenDesk identity services.
+- `govoplan-mail#5`: Open-Xchange mail/groupware adapter boundary.
+- `govoplan-calendar#2`: Open-Xchange calendar adapter boundary.
+- `govoplan-connectors#1`: OpenProject connector.
+
+Future executable connector issues should be created in the owning module
+repository when the first concrete API/profile slice is selected.

docs/OPENPROJECT_CONNECTOR.md

@@ -0,0 +1,170 @@
+# OpenProject Connector Concept
+
+OpenProject is the first proposed concrete connector for
+`govoplan-connectors`. It gives GovOPlaN a public-sector-friendly project and
+work-package integration target without making project management a core
+platform dependency.
+
+## Goals
+
+- Register OpenProject connection profiles.
+- Test API reachability and authentication without exposing secrets.
+- Read projects, users, statuses, and work packages for linking.
+- Create or update work packages from GovOPlaN tasks/cases/workflows once those
+  modules request the capability.
+- Receive or poll changes for external-reference synchronization.
+- Keep all OpenProject-specific client code inside the connector module.
+
+## Non-Goals
+
+- Replacing a future native GovOPlaN project-management module.
+- Importing workflow, tasks, cases, or access implementation modules directly.
+- Mirroring complete OpenProject project state into GovOPlaN by default.
+- Storing OpenProject tokens outside the platform secret contract.
+
+## Profile Fields
+
+Candidate profile payload:
+
+- `base_url`
+- `api_version`, default `v3`
+- `credential_ref`
+- `verify_tls`
+- `timeout_seconds`
+- `allowed_project_ids`
+- `default_project_id`
+- `webhook_secret_ref`, optional
+- `poll_interval_seconds`, optional
+
+## Health Check
+
+The connection test should:
+
+1. Normalize and validate `base_url`.
+2. Resolve the credential reference.
+3. Call the OpenProject API root or a small read-only endpoint.
+4. Record API version, authenticated principal where available, latency,
+   response status, and safe capability hints.
+5. Redact token, Authorization headers, cookies, and any server-provided secret
+   fields from diagnostics.
+
+## Candidate Capabilities
+
+- `connectors.openproject.profileTester`
+- `connectors.openproject.projects`
+- `connectors.openproject.workPackages.read`
+- `connectors.openproject.workPackages.write`
+- `connectors.openproject.webhooks`
+- `connectors.openproject.externalReferences`
+
+Domain modules request these through core-mediated capabilities. For example,
+`govoplan-tasks` can publish a task as an OpenProject work package without
+importing OpenProject client code.
+
+## Data Boundary Decision
+
+OpenProject should be referenced live by default, not mirrored wholesale into
+GovOPlaN. The connector stores stable external references and safe metadata:
+
+- profile id and connector type
+- project id and work-package id
+- external URL
+- remote version, ETag, or lock version where available
+- last-seen timestamp and safe status/type labels
+- GovOPlaN trace id for publish or synchronization jobs
+
+GovOPlaN should import only the subset needed by a requesting domain module,
+for example a work-package title/status for display, a link-back reference for a
+task, or evidence that a publish operation succeeded. Full project state,
+comments, attachments, membership lists, and custom fields remain remote unless
+a future domain module explicitly owns that synchronization. This keeps cases,
+tasks, workflow, and reporting decoupled from OpenProject while still allowing
+link-out, link-back, selected publish, and selected read views.
+
+## Runtime Events
+
+- `openproject.profile_tested`
+- `openproject.project_seen`
+- `openproject.work_package_seen`
+- `openproject.work_package_published`
+- `openproject.webhook_received`
+- `openproject.sync_failed`
+
+Events should carry GovOPlaN ids, external ids, safe diagnostics, and trace
+context. They must not contain credentials or raw personal data beyond what the
+requesting domain module is authorized to process.
+
+## Webhook And Polling Strategy
+
+OpenProject supports API and webhook administration. The connector should allow
+both:
+
+- Webhook-first when an operator registers a webhook for selected project/work
+  package events.
+- Polling fallback for installations where webhooks cannot be exposed.
+
+The first implementation can start with manual test plus read-only project/work
+package lookup, then add publishing, then webhook/polling synchronization.
+
+## First Implementation Slice
+
+1. Add connector type metadata for `openproject`.
+2. Add connection profile CRUD using secret references.
+3. Add a read-only test endpoint.
+4. Add project/work-package lookup DTOs.
+5. Add external-reference storage for linked OpenProject work packages.
+6. Add a WebUI profile page with last-test diagnostics.
+7. Add tests for redaction, unavailable connector behavior, and optional module
+   capability discovery.
+
+## Minimum DTOs
+
+Profile summary:
+
+```json
+{
+  "id": "openproject-main",
+  "connector_type": "openproject",
+  "name": "OpenProject",
+  "base_url": "https://openproject.example",
+  "state": "tested",
+  "last_test_at": "2026-07-09T10:00:00Z",
+  "last_test_status": "success"
+}
+```
+
+External reference:
+
+```json
+{
+  "connector_type": "openproject",
+  "profile_id": "openproject-main",
+  "object_type": "work_package",
+  "external_id": "1234",
+  "external_url": "https://openproject.example/work_packages/1234",
+  "version": "etag-or-lock-version",
+  "metadata": {
+    "project_id": "42"
+  }
+}
+```
+
+Diagnostics must be redacted and should include only endpoint, version,
+authenticated principal label where safe, latency, status code, and capability
+hints.
+
+## First Tests To Add
+
+- profile create/update rejects plaintext token fields
+- profile test redacts Authorization, cookies, and token-like response fields
+- lookup capabilities are absent when the connector module is disabled
+- a task/workflow/case module can detect OpenProject capabilities without
+  importing connector internals
+- external-reference round-trip stores profile id, object type, external id,
+  version/ETag, URL, and safe metadata
+
+## Reference Sources
+
+- OpenProject API v3 documentation: https://www.openproject.org/docs/api/
+- OpenProject API introduction: https://www.openproject.org/docs/api/introduction/
+- OpenProject API and webhooks administration: https://www.openproject.org/docs/system-admin-guide/api-and-webhooks/

docs/PUBLIC_SECTOR_INTEGRATION_CATALOGUE.md

@@ -0,0 +1,156 @@
+# Public-Sector Integration Catalogue
+
+`govoplan-connectors` should maintain an operator-visible catalogue of common
+external systems, protocols, and integration patterns. The catalogue is not a
+promise that GovOPlaN replaces those systems. It is the map that lets modules
+discover what exists, test connections, and decide which optional behavior can
+be enabled.
+
+## Catalogue Entry Shape
+
+Each connector type should define:
+
+- stable connector type key, for example `openproject`, `fit-connect`,
+  `xrepository`, or `sap`
+- category and owning GovOPlaN module, if any
+- supported direction: consume, publish, or bidirectional
+- supported trigger modes: manual test, polling, webhook, batch import, export
+- credential method and whether secrets are stored through the platform secret
+  contract
+- health check and diagnostic payload shape
+- external reference shape for records created or linked through the connector
+- required capabilities and optional module combinations
+- lifecycle support: activate, pause, re-test, rotate credential, retire
+
+## Initial Target Categories
+
+The catalogue should be maintained as a ranked inventory. A target can start as
+an inventory entry before there is executable connector code.
+
+| Target | Scope/Jurisdiction | Category | Mode | Likely Owner | First Useful Capability | Priority |
+| --- | --- | --- | --- | --- | --- | --- |
+| OpenProject | international/open source | Project/task management | link, synchronize selected records, publish tasks | `govoplan-connectors`, later tasks/workflow/projects | profile test, project/work-package lookup, external references | Wave 0 |
+| Nextcloud/WebDAV/SMB/Seafile | broad public-sector/self-hosted | File providers | integrate, import, link | `govoplan-files` with connector inventory | profile health and managed-file provenance | Wave 0/in progress |
+| OpenDesk IDM, LDAP, Active Directory, OIDC, SAML | Germany/EU and general enterprise | Identity | integrate, synchronize | `govoplan-idm`, `govoplan-access` | endpoint inventory, login/provisioning preflight | Wave 1 |
+| Open-Xchange mail/calendar | Germany/OpenDesk and groupware deployments | Groupware | integrate, link | `govoplan-mail`, `govoplan-calendar` | profile test, mailbox/calendar diagnostics | Wave 1 |
+| FIT-Connect | German public-sector transport | Public-sector transport | integrate, publish, receive | dedicated protocol module with connectors inventory | destination profile, test, receipt reference | Wave 1 |
+| XRepository/XÖV lookup | German public-sector standards | Standards registry | link, import schema metadata | `govoplan-connectors`, later XÖV modules | read-only catalogue lookup/cache | Wave 1 |
+| RSS/API publication | public data/external services | Publication/data exchange | consume, publish | `govoplan-connectors`, `govoplan-dataflow` | consume/publish feed profiles | Wave 2 |
+| DMS/e-file/archive systems | German municipal/state/federal administration | DMS/records | link, import, synchronize selected metadata | `govoplan-dms`, `govoplan-files` | external document reference and health | Wave 2 |
+| ERP/finance/procurement/payment systems | German municipal finance/procurement plus EU standards | ERP/payment | export, import, synchronize, replace only by domain decision | dedicated modules | profile inventory and export/import staging | Wave 2 |
+
+### Project And Task Management
+
+- OpenProject
+- Jira or Jira-compatible APIs
+- Redmine
+- Microsoft Planner/Project where available through Microsoft Graph
+
+GovOPlaN should start with OpenProject because it is open source, common in
+public-sector environments, and has API/webhook documentation suitable for a
+first connector.
+
+### DMS, E-File, Records, And Archive
+
+- d.velop/d.3
+- enaio
+- Fabasoft eGov-Suite
+- ELO
+- VIS/eAkte environments
+- CMIS-capable repositories
+- S3/object storage used as archive staging
+
+These targets should usually be owned by DMS/files/records modules once a
+domain module exists. `govoplan-connectors` should still provide inventory,
+profiles, and generic health checks.
+
+### File Providers
+
+- SMB/CIFS
+- WebDAV
+- Nextcloud
+- Seafile
+- S3-compatible object storage
+- SFTP
+
+The files module owns file semantics. The connectors catalogue should record
+profile metadata and health, but must not import files-module internals.
+
+### Identity And Access
+
+- LDAP
+- Active Directory
+- OIDC
+- SAML
+- OpenDesk IDM and comparable identity platforms
+
+The access/IDM modules own principal synchronization and authorization effects.
+Connectors own endpoint inventory and diagnostics.
+
+### Mail, Calendar, And Collaboration
+
+- Microsoft Exchange/M365
+- Open-Xchange
+- IMAP/SMTP where represented as external infrastructure
+- CalDAV/CardDAV
+- chat, video, and collaboration systems such as Matrix, Jitsi, BigBlueButton,
+  Nextcloud Talk, or Collabora/OnlyOffice environments
+
+Mail/calendar/collaboration modules own business semantics. Connector profiles
+can expose reachability and version diagnostics.
+
+### ERP, Finance, Procurement, And Payment
+
+- SAP
+- MACH
+- Infoma/new system
+- DATEV interfaces
+- XRechnung/Peppol access points
+- XBestellung and procurement feeds
+- payment providers and cash-register systems
+
+Protocol-heavy parts should move into dedicated modules such as
+`govoplan-xrechnung`, `govoplan-erp`, `govoplan-procurement`, or
+`govoplan-payments`.
+
+### Public-Sector Protocols And Registries
+
+- FIT-Connect
+- XTA/OSCI
+- XÖV standards and XRepository lookup
+- XRechnung/XBestellung
+- register and Fachverfahren interfaces discovered by implementation projects
+
+These are integration priorities because they model common administrative
+processes. They should be represented as connector categories even when a
+dedicated module later owns the actual protocol implementation.
+
+## Wave 0 Catalogue Priorities
+
+1. OpenProject connector concept and profile shape.
+2. Generic connector profile, health, and secret-reference model.
+3. Public-sector target inventory table with category, owner module, and
+   priority.
+4. Consume/publish source lifecycle contract.
+5. External-reference DTO shared through kernel/core contracts.
+6. Configuration-package declaration for required connector profiles.
+
+## Catalogue Maintenance Rules
+
+- Prefer one stable connector type key per external product or protocol family.
+- Record when GovOPlaN should integrate with an existing product instead of
+  replacing it.
+- Keep protocol/client implementation in the owning connector or protocol
+  module; domain modules consume capabilities and DTOs only.
+- Treat "inventory only" entries as useful: operators can document a landscape
+  before GovOPlaN can automate it.
+- Every executable connector type needs a redaction-safe test plan, lifecycle
+  states, external-reference shape, and uninstall/retirement behavior.
+
+## Reference Sources
+
+- OpenProject API v3 documentation: https://www.openproject.org/docs/api/
+- OpenProject API and webhooks administration: https://www.openproject.org/docs/system-admin-guide/api-and-webhooks/
+- FIT-Connect Destination API documentation: https://docs.fitko.de/en/resources/fit-connect-destination-api/
+- XÖV overview by KoSIT: https://www.xoev.de/xoev-4987
+- XRepository overview: https://www.xrepository.de/

pyproject.toml

@@ -0,0 +1,22 @@
+[build-system]
+requires = ["setuptools>=69", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "govoplan-connectors"
+version = "0.1.14"
+description = "Governed connector catalogue and tabular source capabilities for GovOPlaN."
+readme = "README.md"
+requires-python = ">=3.12"
+license = "AGPL-3.0-or-later"
+authors = [{ name = "GovOPlaN" }]
+dependencies = ["govoplan-core>=0.1.14"]
+
+[tool.setuptools.packages.find]
+where = ["src"]
+
+[tool.setuptools.package-data]
+govoplan_connectors = ["py.typed"]
+
+[project.entry-points."govoplan.modules"]
+connectors = "govoplan_connectors.backend.manifest:get_manifest"

src/govoplan_connectors/__init__.py

@@ -0,0 +1,3 @@
+"""GovOPlaN Connectors module."""
+
+__all__: list[str] = []

src/govoplan_connectors/backend/__init__.py

@@ -0,0 +1 @@
+"""Connector backend package."""

src/govoplan_connectors/backend/datasource_origins.py

@@ -0,0 +1,132 @@
+from __future__ import annotations
+
+from govoplan_core.core.datasources import (
+    DatasourceAccessError,
+    DatasourceField,
+    DatasourceNotFoundError,
+    DatasourceOrigin,
+    DatasourceOriginReadRequest,
+    DatasourceOriginReadResult,
+    DatasourceValidationError,
+)
+from govoplan_core.core.tabular_sources import (
+    TabularReadRequest,
+    TabularSource,
+    TabularSourceAccessError,
+    TabularSourceError,
+    TabularSourceNotFoundError,
+    TabularSourceValidationError,
+)
+from govoplan_connectors.backend.tabular_sources import SqlTabularSourceProvider
+
+
+class ConnectorDatasourceOriginProvider:
+    """Expose connector-owned sources through the Datasources origin contract."""
+
+    def __init__(self, provider: SqlTabularSourceProvider | None = None) -> None:
+        self._provider = provider or SqlTabularSourceProvider()
+
+    def list_origins(
+        self,
+        session: object,
+        principal: object,
+        *,
+        query: str = "",
+        limit: int = 100,
+    ):
+        try:
+            rows = self._provider.list_sources(
+                session,
+                principal,
+                query=query,
+                limit=limit,
+            )
+        except TabularSourceError as exc:
+            raise _datasource_error(exc) from exc
+        return tuple(_origin(source) for source in rows)
+
+    def get_origin(
+        self,
+        session: object,
+        principal: object,
+        *,
+        origin_ref: str,
+    ) -> DatasourceOrigin | None:
+        try:
+            source = self._provider.get_source(
+                session,
+                principal,
+                source_ref=origin_ref,
+            )
+        except TabularSourceError as exc:
+            raise _datasource_error(exc) from exc
+        return _origin(source) if source is not None else None
+
+    def read_origin(
+        self,
+        session: object,
+        principal: object,
+        *,
+        request: DatasourceOriginReadRequest,
+    ) -> DatasourceOriginReadResult:
+        try:
+            result = self._provider.read_source(
+                session,
+                principal,
+                request=TabularReadRequest(
+                    source_ref=request.origin_ref,
+                    limit=request.limit,
+                    offset=request.offset,
+                    columns=request.columns,
+                    expected_fingerprint=request.expected_fingerprint,
+                ),
+            )
+        except TabularSourceError as exc:
+            raise _datasource_error(exc) from exc
+        return DatasourceOriginReadResult(
+            origin=_origin(result.source),
+            rows=result.rows,
+            total_rows=result.total_rows,
+            truncated=result.truncated,
+        )
+
+
+def _origin(source: TabularSource) -> DatasourceOrigin:
+    return DatasourceOrigin(
+        ref=source.ref,
+        source_name=source.source_name,
+        name=source.name,
+        description=source.description,
+        kind="upload",
+        shape="tabular",
+        supported_modes=("live", "cached"),
+        provider=f"connectors.{source.provider}",
+        schema=tuple(
+            DatasourceField(
+                name=column.name,
+                data_type=column.data_type,
+                nullable=column.nullable,
+            )
+            for column in source.schema
+        ),
+        schema_version=source.schema_version,
+        fingerprint=source.fingerprint,
+        row_count=source.row_count,
+        byte_count=source.byte_count,
+        updated_at=source.updated_at,
+        capabilities=source.capabilities,
+        metadata=dict(source.metadata),
+    )
+
+
+def _datasource_error(exc: TabularSourceError):
+    if isinstance(exc, TabularSourceAccessError):
+        return DatasourceAccessError(str(exc))
+    if isinstance(exc, TabularSourceNotFoundError):
+        return DatasourceNotFoundError(str(exc))
+    if isinstance(exc, TabularSourceValidationError):
+        return DatasourceValidationError(str(exc))
+    return DatasourceValidationError(str(exc))
+
+
+__all__ = ["ConnectorDatasourceOriginProvider"]

src/govoplan_connectors/backend/db/__init__.py

@@ -0,0 +1,3 @@
+from govoplan_connectors.backend.db.models import ConnectorTabularSource
+
+__all__ = ["ConnectorTabularSource"]

src/govoplan_connectors/backend/db/models.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+from typing import Any
+
+from sqlalchemy import DateTime, Index, Integer, JSON, String, Text, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from govoplan_core.db.base import Base, TimestampMixin
+
+
+def new_uuid() -> str:
+    return str(uuid.uuid4())
+
+
+class ConnectorTabularSource(Base, TimestampMixin):
+    __tablename__ = "connector_tabular_sources"
+    __table_args__ = (
+        UniqueConstraint("tenant_id", "source_name", name="uq_connector_tabular_source_name"),
+        Index("ix_connector_tabular_sources_tenant_status", "tenant_id", "status"),
+        Index("ix_connector_tabular_sources_tenant_updated", "tenant_id", "updated_at"),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
+    provider: Mapped[str] = mapped_column(String(50), default="snapshot", nullable=False, index=True)
+    source_name: Mapped[str] = mapped_column(String(120), nullable=False)
+    name: Mapped[str] = mapped_column(String(300), nullable=False)
+    description: Mapped[str | None] = mapped_column(Text)
+    status: Mapped[str] = mapped_column(String(30), default="active", nullable=False, index=True)
+    schema_version: Mapped[int] = mapped_column(Integer, default=1, nullable=False)
+    schema_: Mapped[list[dict[str, Any]]] = mapped_column("schema", JSON, default=list, nullable=False)
+    rows: Mapped[list[dict[str, Any]]] = mapped_column(JSON, default=list, nullable=False)
+    fingerprint: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    row_count: Mapped[int] = mapped_column(Integer, nullable=False)
+    byte_count: Mapped[int] = mapped_column(Integer, nullable=False)
+    metadata_: Mapped[dict[str, Any]] = mapped_column("metadata", JSON, default=dict, nullable=False)
+    created_by: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
+    updated_by: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
+    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
+
+
+__all__ = ["ConnectorTabularSource", "new_uuid"]

src/govoplan_connectors/backend/manifest.py

@@ -0,0 +1,206 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from govoplan_core.core.access import (
+    CAPABILITY_AUTH_PERMISSION_EVALUATOR,
+    CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
+)
+from govoplan_core.core.module_guards import (
+    drop_table_retirement_provider,
+    persistent_table_uninstall_guard,
+)
+from govoplan_core.core.datasources import CAPABILITY_DATASOURCE_ORIGINS
+from govoplan_core.core.modules import (
+    DocumentationTopic,
+    MigrationSpec,
+    ModuleInterfaceProvider,
+    ModuleManifest,
+    PermissionDefinition,
+    RoleTemplate,
+)
+from govoplan_core.core.tabular_sources import (
+    CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER,
+    CAPABILITY_CONNECTORS_TABULAR_SOURCES,
+)
+from govoplan_core.db.base import Base
+from govoplan_connectors.backend.db.models import ConnectorTabularSource
+from govoplan_connectors.backend.tabular_sources import (
+    ADMIN_SCOPE,
+    READ_SCOPE,
+    WRITE_SCOPE,
+    SqlTabularSourceProvider,
+)
+from govoplan_connectors.backend.datasource_origins import (
+    ConnectorDatasourceOriginProvider,
+)
+
+
+MODULE_ID = "connectors"
+MODULE_VERSION = "0.1.14"
+TABULAR_SOURCE_INTERFACE_VERSION = "0.1.0"
+DATASOURCE_ORIGIN_INTERFACE_VERSION = "0.1.0"
+
+
+def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
+    module_id, resource, action = scope.split(":", 2)
+    return PermissionDefinition(
+        scope=scope,
+        label=label,
+        description=description,
+        category="Connectors",
+        level="tenant",
+        module_id=module_id,
+        resource=resource,
+        action=action,
+    )
+
+
+PERMISSIONS = (
+    _permission(
+        READ_SCOPE,
+        "View tabular sources",
+        "Discover and preview policy-visible tabular connector sources.",
+    ),
+    _permission(
+        WRITE_SCOPE,
+        "Manage tabular sources",
+        "Import and retire bounded tabular snapshots.",
+    ),
+    _permission(
+        ADMIN_SCOPE,
+        "Administer connector sources",
+        "Manage every tenant connector source and future source policies.",
+    ),
+)
+
+ROLE_TEMPLATES = (
+    RoleTemplate(
+        slug="connector_source_manager",
+        name="Connector source manager",
+        description="Discover, import, preview, and retire tabular sources.",
+        permissions=(READ_SCOPE, WRITE_SCOPE),
+    ),
+    RoleTemplate(
+        slug="connector_source_reader",
+        name="Connector source reader",
+        description="Discover and preview tabular connector sources.",
+        permissions=(READ_SCOPE,),
+    ),
+)
+
+
+def _router(_context):
+    from govoplan_connectors.backend.router import router
+
+    return router
+
+
+def _provider(_context) -> SqlTabularSourceProvider:
+    return SqlTabularSourceProvider()
+
+
+def _datasource_origin_provider(_context) -> ConnectorDatasourceOriginProvider:
+    return ConnectorDatasourceOriginProvider()
+
+
+def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
+    return {
+        "connector_tabular_sources": (
+            session.query(ConnectorTabularSource)
+            .filter(
+                ConnectorTabularSource.tenant_id == tenant_id,
+                ConnectorTabularSource.deleted_at.is_(None),
+            )
+            .count()
+        )
+    }
+
+
+manifest = ModuleManifest(
+    id=MODULE_ID,
+    name="Connectors",
+    version=MODULE_VERSION,
+    optional_dependencies=("access", "audit", "files", "policy"),
+    required_capabilities=(
+        CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
+        CAPABILITY_AUTH_PERMISSION_EVALUATOR,
+    ),
+    provides_interfaces=(
+        ModuleInterfaceProvider(
+            name="connectors.tabular_sources",
+            version=TABULAR_SOURCE_INTERFACE_VERSION,
+        ),
+        ModuleInterfaceProvider(
+            name="connectors.tabular_snapshot_writer",
+            version=TABULAR_SOURCE_INTERFACE_VERSION,
+        ),
+        ModuleInterfaceProvider(
+            name="connectors.datasource_origins",
+            version=DATASOURCE_ORIGIN_INTERFACE_VERSION,
+        ),
+    ),
+    permissions=PERMISSIONS,
+    role_templates=ROLE_TEMPLATES,
+    route_factory=_router,
+    capability_factories={
+        CAPABILITY_CONNECTORS_TABULAR_SOURCES: _provider,
+        CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER: _provider,
+        CAPABILITY_DATASOURCE_ORIGINS: _datasource_origin_provider,
+    },
+    tenant_summary_providers=(_tenant_summary,),
+    migration_spec=MigrationSpec(
+        module_id=MODULE_ID,
+        metadata=Base.metadata,
+        script_location=str(Path(__file__).with_name("migrations") / "versions"),
+        retirement_supported=True,
+        retirement_provider=drop_table_retirement_provider(
+            ConnectorTabularSource,
+            label="Connectors",
+        ),
+        retirement_notes=(
+            "Destructive retirement drops connector-owned source snapshots after "
+            "the installer captures a database snapshot."
+        ),
+    ),
+    uninstall_guard_providers=(
+        persistent_table_uninstall_guard(
+            ConnectorTabularSource,
+            label="Connectors",
+        ),
+    ),
+    documentation=(
+        DocumentationTopic(
+            id="connectors.tabular-sources",
+            title="Governed tabular sources",
+            summary="Provider-neutral source discovery and bounded reads for Dataflow.",
+            body=(
+                "Connectors owns source configuration, access checks, schema discovery, "
+                "fingerprints, and bounded reads. Dataflow stores only opaque source "
+                "references and expected fingerprints. The first executable provider "
+                "imports immutable JSON or CSV snapshots and exposes them as Datasource "
+                "origins. Database and API providers can implement the same origin "
+                "contract without changing Datasources or Dataflow."
+            ),
+            layer="available",
+            documentation_types=("admin", "user"),
+            audience=("operator", "module_admin", "power_user"),
+            related_modules=("dataflow", "files", "reporting", "risk_compliance"),
+            order=40,
+        ),
+    ),
+)
+
+
+def get_manifest() -> ModuleManifest:
+    return manifest
+
+
+__all__ = [
+    "MODULE_ID",
+    "MODULE_VERSION",
+    "DATASOURCE_ORIGIN_INTERFACE_VERSION",
+    "TABULAR_SOURCE_INTERFACE_VERSION",
+    "get_manifest",
+    "manifest",
+]

src/govoplan_connectors/backend/migrations/__init__.py

@@ -0,0 +1 @@
+"""Connectors migrations."""

src/govoplan_connectors/backend/migrations/versions/__init__.py

@@ -0,0 +1 @@
+"""Connectors migration revisions."""

src/govoplan_connectors/backend/migrations/versions/e6b7c8d9f0a1_v0114_connectors_baseline.py

@@ -0,0 +1,141 @@
+"""v0.1.14 Connectors baseline
+
+Revision ID: e6b7c8d9f0a1
+Revises: None
+Create Date: 2026-07-28 00:00:00.000000
+"""
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "e6b7c8d9f0a1"
+down_revision = None
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "connector_tabular_sources",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=36), nullable=False),
+        sa.Column("provider", sa.String(length=50), nullable=False),
+        sa.Column("source_name", sa.String(length=120), nullable=False),
+        sa.Column("name", sa.String(length=300), nullable=False),
+        sa.Column("description", sa.Text(), nullable=True),
+        sa.Column("status", sa.String(length=30), nullable=False),
+        sa.Column("schema_version", sa.Integer(), nullable=False),
+        sa.Column("schema", sa.JSON(), nullable=False),
+        sa.Column("rows", sa.JSON(), nullable=False),
+        sa.Column("fingerprint", sa.String(length=64), nullable=False),
+        sa.Column("row_count", sa.Integer(), nullable=False),
+        sa.Column("byte_count", sa.Integer(), nullable=False),
+        sa.Column("metadata", sa.JSON(), nullable=False),
+        sa.Column("created_by", sa.String(length=255), nullable=True),
+        sa.Column("updated_by", sa.String(length=255), nullable=True),
+        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_connector_tabular_sources")),
+        sa.UniqueConstraint(
+            "tenant_id",
+            "source_name",
+            name="uq_connector_tabular_source_name",
+        ),
+    )
+    op.create_index(
+        op.f("ix_connector_tabular_sources_created_by"),
+        "connector_tabular_sources",
+        ["created_by"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_connector_tabular_sources_deleted_at"),
+        "connector_tabular_sources",
+        ["deleted_at"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_connector_tabular_sources_fingerprint"),
+        "connector_tabular_sources",
+        ["fingerprint"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_connector_tabular_sources_provider"),
+        "connector_tabular_sources",
+        ["provider"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_connector_tabular_sources_status"),
+        "connector_tabular_sources",
+        ["status"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_connector_tabular_sources_tenant_id"),
+        "connector_tabular_sources",
+        ["tenant_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_connector_tabular_sources_updated_by"),
+        "connector_tabular_sources",
+        ["updated_by"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_connector_tabular_sources_tenant_status",
+        "connector_tabular_sources",
+        ["tenant_id", "status"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_connector_tabular_sources_tenant_updated",
+        "connector_tabular_sources",
+        ["tenant_id", "updated_at"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "ix_connector_tabular_sources_tenant_updated",
+        table_name="connector_tabular_sources",
+    )
+    op.drop_index(
+        "ix_connector_tabular_sources_tenant_status",
+        table_name="connector_tabular_sources",
+    )
+    op.drop_index(
+        op.f("ix_connector_tabular_sources_updated_by"),
+        table_name="connector_tabular_sources",
+    )
+    op.drop_index(
+        op.f("ix_connector_tabular_sources_tenant_id"),
+        table_name="connector_tabular_sources",
+    )
+    op.drop_index(
+        op.f("ix_connector_tabular_sources_status"),
+        table_name="connector_tabular_sources",
+    )
+    op.drop_index(
+        op.f("ix_connector_tabular_sources_provider"),
+        table_name="connector_tabular_sources",
+    )
+    op.drop_index(
+        op.f("ix_connector_tabular_sources_fingerprint"),
+        table_name="connector_tabular_sources",
+    )
+    op.drop_index(
+        op.f("ix_connector_tabular_sources_deleted_at"),
+        table_name="connector_tabular_sources",
+    )
+    op.drop_index(
+        op.f("ix_connector_tabular_sources_created_by"),
+        table_name="connector_tabular_sources",
+    )
+    op.drop_table("connector_tabular_sources")

src/govoplan_connectors/backend/router.py

@@ -0,0 +1,211 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlalchemy.orm import Session
+
+from govoplan_core.audit.logging import audit_event
+from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
+from govoplan_core.core.tabular_sources import (
+    TabularReadRequest,
+    TabularSnapshotInput,
+    TabularSource,
+    TabularSourceAccessError,
+    TabularSourceError,
+    TabularSourceNotFoundError,
+)
+from govoplan_core.db.session import get_session
+from govoplan_connectors.backend.schemas import (
+    SnapshotCreateRequest,
+    TabularColumnResponse,
+    TabularSourceDeleteResponse,
+    TabularSourceListResponse,
+    TabularSourcePreviewResponse,
+    TabularSourceResponse,
+)
+from govoplan_connectors.backend.tabular_sources import (
+    ADMIN_SCOPE,
+    READ_SCOPE,
+    WRITE_SCOPE,
+    SqlTabularSourceProvider,
+    parse_csv_snapshot,
+)
+
+
+router = APIRouter(prefix="/connectors", tags=["connectors"])
+provider = SqlTabularSourceProvider()
+
+
+def _require_any_scope(principal: ApiPrincipal, *scopes: str) -> None:
+    if any(has_scope(principal, scope) for scope in scopes):
+        return
+    raise HTTPException(
+        status_code=status.HTTP_403_FORBIDDEN,
+        detail=f"Missing one of the required scopes: {', '.join(scopes)}",
+    )
+
+
+def _http_error(exc: TabularSourceError) -> HTTPException:
+    if isinstance(exc, TabularSourceNotFoundError):
+        return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
+    if isinstance(exc, TabularSourceAccessError):
+        return HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc))
+    return HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc))
+
+
+@router.get("/tabular-sources", response_model=TabularSourceListResponse)
+def api_list_tabular_sources(
+    query: str = Query(default="", max_length=200),
+    limit: int = Query(default=100, ge=1, le=100),
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> TabularSourceListResponse:
+    _require_any_scope(principal, READ_SCOPE, ADMIN_SCOPE)
+    sources = provider.list_sources(
+        session,
+        principal,
+        query=query,
+        limit=limit,
+    )
+    return TabularSourceListResponse(sources=[_source_response(source) for source in sources])
+
+
+@router.post(
+    "/tabular-sources/snapshots",
+    response_model=TabularSourceResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+def api_create_tabular_snapshot(
+    payload: SnapshotCreateRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> TabularSourceResponse:
+    _require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
+    try:
+        rows = (
+            tuple(payload.rows or ())
+            if payload.format == "json"
+            else parse_csv_snapshot(payload.csv_text or "", delimiter=payload.delimiter)
+        )
+        source = provider.create_snapshot(
+            session,
+            principal,
+            snapshot=TabularSnapshotInput(
+                name=payload.name,
+                source_name=payload.source_name,
+                description=payload.description,
+                rows=rows,
+                metadata={"import_format": payload.format},
+            ),
+        )
+    except TabularSourceError as exc:
+        raise _http_error(exc) from exc
+    audit_event(
+        session,
+        tenant_id=principal.tenant_id,
+        user_id=getattr(principal.user, "id", None),
+        api_key_id=principal.api_key_id,
+        action="connectors.tabular_snapshot.created",
+        object_type="connector_tabular_source",
+        object_id=source.ref,
+        details={
+            "provider": source.provider,
+            "source_name": source.source_name,
+            "fingerprint": source.fingerprint,
+            "row_count": source.row_count,
+        },
+    )
+    session.commit()
+    return _source_response(source)
+
+
+@router.get(
+    "/tabular-sources/{source_id}/preview",
+    response_model=TabularSourcePreviewResponse,
+)
+def api_preview_tabular_source(
+    source_id: str,
+    limit: int = Query(default=100, ge=1, le=500),
+    offset: int = Query(default=0, ge=0),
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> TabularSourcePreviewResponse:
+    _require_any_scope(principal, READ_SCOPE, ADMIN_SCOPE)
+    try:
+        result = provider.read_source(
+            session,
+            principal,
+            request=TabularReadRequest(
+                source_ref=f"snapshot:{source_id}",
+                limit=limit,
+                offset=offset,
+            ),
+        )
+    except TabularSourceError as exc:
+        raise _http_error(exc) from exc
+    return TabularSourcePreviewResponse(
+        source=_source_response(result.source),
+        rows=[dict(row) for row in result.rows],
+        total_rows=result.total_rows,
+        truncated=result.truncated,
+    )
+
+
+@router.delete(
+    "/tabular-sources/{source_id}",
+    response_model=TabularSourceDeleteResponse,
+)
+def api_delete_tabular_source(
+    source_id: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> TabularSourceDeleteResponse:
+    _require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
+    source_ref = f"snapshot:{source_id}"
+    try:
+        source = provider.delete_snapshot(
+            session,
+            principal,
+            source_ref=source_ref,
+        )
+    except TabularSourceError as exc:
+        raise _http_error(exc) from exc
+    audit_event(
+        session,
+        tenant_id=principal.tenant_id,
+        user_id=getattr(principal.user, "id", None),
+        api_key_id=principal.api_key_id,
+        action="connectors.tabular_snapshot.deleted",
+        object_type="connector_tabular_source",
+        object_id=source_ref,
+        details={"source_name": source.source_name, "fingerprint": source.fingerprint},
+    )
+    session.commit()
+    return TabularSourceDeleteResponse(deleted=True, source_ref=source_ref)
+
+
+def _source_response(source: TabularSource) -> TabularSourceResponse:
+    return TabularSourceResponse(
+        ref=source.ref,
+        provider=source.provider,
+        source_name=source.source_name,
+        name=source.name,
+        description=source.description,
+        columns=[
+            TabularColumnResponse(
+                name=column.name,
+                data_type=column.data_type,
+                nullable=column.nullable,
+            )
+            for column in source.schema
+        ],
+        schema_version=source.schema_version,
+        fingerprint=source.fingerprint,
+        row_count=source.row_count,
+        byte_count=source.byte_count,
+        updated_at=source.updated_at.isoformat() if source.updated_at else None,
+        capabilities=list(source.capabilities),
+        metadata=dict(source.metadata),
+    )
+
+
+__all__ = ["router"]

src/govoplan_connectors/backend/schemas.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field, model_validator
+
+
+class SnapshotCreateRequest(BaseModel):
+    name: str = Field(min_length=1, max_length=300)
+    source_name: str = Field(
+        min_length=1,
+        max_length=120,
+        pattern=r"^[A-Za-z_][A-Za-z0-9_]*$",
+    )
+    description: str | None = Field(default=None, max_length=4000)
+    format: Literal["json", "csv"] = "json"
+    rows: list[dict[str, Any]] | None = Field(default=None, max_length=10_000)
+    csv_text: str | None = Field(default=None, max_length=5_000_000)
+    delimiter: Literal[",", ";", "\t", "|"] = ","
+
+    @model_validator(mode="after")
+    def validate_payload(self) -> "SnapshotCreateRequest":
+        if self.format == "json" and self.rows is None:
+            raise ValueError("JSON snapshots require rows.")
+        if self.format == "json" and self.csv_text is not None:
+            raise ValueError("JSON snapshots cannot include CSV text.")
+        if self.format == "csv" and not self.csv_text:
+            raise ValueError("CSV snapshots require CSV text.")
+        if self.format == "csv" and self.rows is not None:
+            raise ValueError("CSV snapshots cannot include JSON rows.")
+        return self
+
+
+class TabularColumnResponse(BaseModel):
+    name: str
+    data_type: str
+    nullable: bool
+
+
+class TabularSourceResponse(BaseModel):
+    ref: str
+    provider: str
+    source_name: str
+    name: str
+    description: str | None
+    columns: list[TabularColumnResponse]
+    schema_version: str
+    fingerprint: str
+    row_count: int | None
+    byte_count: int | None
+    updated_at: str | None
+    capabilities: list[str]
+    metadata: dict[str, Any]
+
+
+class TabularSourceListResponse(BaseModel):
+    sources: list[TabularSourceResponse]
+
+
+class TabularSourcePreviewResponse(BaseModel):
+    source: TabularSourceResponse
+    rows: list[dict[str, Any]]
+    total_rows: int
+    truncated: bool
+
+
+class TabularSourceDeleteResponse(BaseModel):
+    deleted: bool
+    source_ref: str
+
+
+__all__ = [
+    "SnapshotCreateRequest",
+    "TabularColumnResponse",
+    "TabularSourceDeleteResponse",
+    "TabularSourceListResponse",
+    "TabularSourcePreviewResponse",
+    "TabularSourceResponse",
+]

src/govoplan_connectors/backend/tabular_sources.py

@@ -0,0 +1,374 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from collections.abc import Mapping, Sequence
+from datetime import datetime
+from decimal import Decimal
+from typing import Any
+
+from sqlalchemy import or_, select
+from sqlalchemy.orm import Session
+
+from govoplan_core.auth import ApiPrincipal, has_scope
+from govoplan_core.core.tabular_sources import (
+    TabularColumn,
+    TabularReadRequest,
+    TabularReadResult,
+    TabularSnapshotInput,
+    TabularSource,
+    TabularSourceAccessError,
+    TabularSourceNotFoundError,
+    TabularSourceValidationError,
+    parse_tabular_csv,
+)
+from govoplan_core.db.base import utcnow
+from govoplan_connectors.backend.db.models import ConnectorTabularSource
+
+
+READ_SCOPE = "connectors:source:read"
+WRITE_SCOPE = "connectors:source:write"
+ADMIN_SCOPE = "connectors:source:admin"
+MAX_SNAPSHOT_ROWS = 10_000
+MAX_SNAPSHOT_BYTES = 5_000_000
+MAX_READ_ROWS = 500
+
+
+class SqlTabularSourceProvider:
+    def list_sources(
+        self,
+        session: object,
+        principal: object,
+        *,
+        query: str = "",
+        limit: int = 100,
+    ) -> Sequence[TabularSource]:
+        db, api_principal = _context(session, principal, READ_SCOPE)
+        normalized_query = str(query or "").strip()
+        statement = select(ConnectorTabularSource).where(
+            ConnectorTabularSource.tenant_id == api_principal.tenant_id,
+            ConnectorTabularSource.deleted_at.is_(None),
+            ConnectorTabularSource.status == "active",
+        )
+        if normalized_query:
+            pattern = f"%{_escape_like(normalized_query)}%"
+            statement = statement.where(
+                or_(
+                    ConnectorTabularSource.name.ilike(pattern, escape="\\"),
+                    ConnectorTabularSource.source_name.ilike(pattern, escape="\\"),
+                )
+            )
+        statement = statement.order_by(
+            ConnectorTabularSource.updated_at.desc(),
+            ConnectorTabularSource.name,
+        ).limit(max(1, min(int(limit), 100)))
+        return tuple(_source_dto(item) for item in db.scalars(statement))
+
+    def get_source(
+        self,
+        session: object,
+        principal: object,
+        *,
+        source_ref: str,
+    ) -> TabularSource | None:
+        db, api_principal = _context(session, principal, READ_SCOPE)
+        item = _source_record(
+            db,
+            tenant_id=api_principal.tenant_id,
+            source_ref=source_ref,
+        )
+        return _source_dto(item) if item is not None else None
+
+    def read_source(
+        self,
+        session: object,
+        principal: object,
+        *,
+        request: TabularReadRequest,
+    ) -> TabularReadResult:
+        db, api_principal = _context(session, principal, READ_SCOPE)
+        item = _source_record(
+            db,
+            tenant_id=api_principal.tenant_id,
+            source_ref=request.source_ref,
+        )
+        if item is None:
+            raise TabularSourceNotFoundError("Tabular source not found.")
+        if request.expected_fingerprint and request.expected_fingerprint != item.fingerprint:
+            raise TabularSourceValidationError(
+                "The source fingerprint changed; refresh the source node before running it."
+            )
+
+        limit = max(1, min(int(request.limit), MAX_READ_ROWS))
+        offset = max(0, int(request.offset))
+        selected_columns = tuple(dict.fromkeys(request.columns))
+        known_columns = {column["name"] for column in item.schema_}
+        unknown_columns = [column for column in selected_columns if column not in known_columns]
+        if unknown_columns:
+            raise TabularSourceValidationError(
+                f"Unknown source columns: {', '.join(unknown_columns)}"
+            )
+        window = item.rows[offset : offset + limit]
+        rows = tuple(
+            {
+                key: value
+                for key, value in row.items()
+                if not selected_columns or key in selected_columns
+            }
+            for row in window
+        )
+        return TabularReadResult(
+            source=_source_dto(item),
+            rows=rows,
+            total_rows=item.row_count,
+            truncated=offset + len(rows) < item.row_count,
+        )
+
+    def create_snapshot(
+        self,
+        session: object,
+        principal: object,
+        *,
+        snapshot: TabularSnapshotInput,
+    ) -> TabularSource:
+        db, api_principal = _context(session, principal, WRITE_SCOPE)
+        name = snapshot.name.strip()
+        source_name = snapshot.source_name.strip()
+        if not name:
+            raise TabularSourceValidationError("Snapshot name is required.")
+        if not source_name:
+            raise TabularSourceValidationError("Snapshot source name is required.")
+        if len(snapshot.rows) > MAX_SNAPSHOT_ROWS:
+            raise TabularSourceValidationError(
+                f"Snapshots are limited to {MAX_SNAPSHOT_ROWS:,} rows."
+            )
+        rows = [_json_row(row) for row in snapshot.rows]
+        encoded = json.dumps(rows, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
+        if len(encoded) > MAX_SNAPSHOT_BYTES:
+            raise TabularSourceValidationError(
+                f"Snapshots are limited to {MAX_SNAPSHOT_BYTES // 1_000_000} MB."
+            )
+        existing = db.scalar(
+            select(ConnectorTabularSource.id).where(
+                ConnectorTabularSource.tenant_id == api_principal.tenant_id,
+                ConnectorTabularSource.source_name == source_name,
+            )
+        )
+        if existing is not None:
+            raise TabularSourceValidationError(
+                f"A tabular source named {source_name!r} already exists."
+            )
+        schema = infer_schema(rows)
+        fingerprint = snapshot_fingerprint(rows, schema)
+        actor_id = _actor_id(api_principal)
+        item = ConnectorTabularSource(
+            tenant_id=api_principal.tenant_id,
+            provider="snapshot",
+            source_name=source_name,
+            name=name,
+            description=_clean_optional(snapshot.description),
+            status="active",
+            schema_version=1,
+            schema_=[_column_payload(column) for column in schema],
+            rows=rows,
+            fingerprint=fingerprint,
+            row_count=len(rows),
+            byte_count=len(encoded),
+            metadata_=dict(snapshot.metadata),
+            created_by=actor_id,
+            updated_by=actor_id,
+        )
+        db.add(item)
+        db.flush()
+        return _source_dto(item)
+
+    def delete_snapshot(
+        self,
+        session: object,
+        principal: object,
+        *,
+        source_ref: str,
+    ) -> TabularSource:
+        db, api_principal = _context(session, principal, WRITE_SCOPE)
+        item = _source_record(
+            db,
+            tenant_id=api_principal.tenant_id,
+            source_ref=source_ref,
+        )
+        if item is None:
+            raise TabularSourceNotFoundError("Tabular source not found.")
+        item.deleted_at = utcnow()
+        item.status = "retired"
+        item.updated_by = _actor_id(api_principal)
+        db.flush()
+        return _source_dto(item)
+
+
+def parse_csv_snapshot(csv_text: str, *, delimiter: str) -> tuple[Mapping[str, object], ...]:
+    return parse_tabular_csv(
+        csv_text,
+        delimiter=delimiter,
+        max_rows=MAX_SNAPSHOT_ROWS,
+    )
+
+
+def infer_schema(rows: Sequence[Mapping[str, object]]) -> tuple[TabularColumn, ...]:
+    names: list[str] = []
+    for row in rows:
+        for name in row:
+            if name not in names:
+                names.append(name)
+    result: list[TabularColumn] = []
+    for name in names:
+        values = [row.get(name) for row in rows]
+        concrete = [value for value in values if value is not None]
+        data_type = _type_name(concrete[0]) if concrete else "unknown"
+        if any(_type_name(value) != data_type for value in concrete[1:]):
+            data_type = "mixed"
+        result.append(
+            TabularColumn(
+                name=name,
+                data_type=data_type,
+                nullable=len(concrete) != len(values),
+            )
+        )
+    return tuple(result)
+
+
+def snapshot_fingerprint(
+    rows: Sequence[Mapping[str, object]],
+    schema: Sequence[TabularColumn],
+) -> str:
+    payload = {
+        "schema": [_column_payload(column) for column in schema],
+        "rows": [dict(row) for row in rows],
+    }
+    encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str)
+    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+
+
+def _source_record(
+    session: Session,
+    *,
+    tenant_id: str,
+    source_ref: str,
+) -> ConnectorTabularSource | None:
+    source_id = source_ref.removeprefix("snapshot:")
+    if not source_id or source_id == source_ref:
+        return None
+    return session.scalar(
+        select(ConnectorTabularSource).where(
+            ConnectorTabularSource.id == source_id,
+            ConnectorTabularSource.tenant_id == tenant_id,
+            ConnectorTabularSource.deleted_at.is_(None),
+        )
+    )
+
+
+def _source_dto(item: ConnectorTabularSource) -> TabularSource:
+    return TabularSource(
+        ref=f"snapshot:{item.id}",
+        provider=item.provider,
+        source_name=item.source_name,
+        name=item.name,
+        description=item.description,
+        schema=tuple(TabularColumn(**column) for column in item.schema_),
+        schema_version=str(item.schema_version),
+        fingerprint=item.fingerprint,
+        row_count=item.row_count,
+        byte_count=item.byte_count,
+        updated_at=item.updated_at,
+        capabilities=("read", "preview"),
+        metadata=dict(item.metadata_),
+    )
+
+
+def _context(
+    session: object,
+    principal: object,
+    required_scope: str,
+) -> tuple[Session, ApiPrincipal]:
+    if not isinstance(session, Session):
+        raise TypeError("Tabular source providers require a SQLAlchemy session.")
+    if not isinstance(principal, ApiPrincipal):
+        raise TabularSourceAccessError("A tenant API principal is required.")
+    accepted_scopes = {required_scope, ADMIN_SCOPE}
+    if required_scope == READ_SCOPE:
+        accepted_scopes.update(
+            {
+                "datasources:catalogue:read",
+                "datasources:source:write",
+                "datasources:source:admin",
+            }
+        )
+    if not any(has_scope(principal, scope) for scope in accepted_scopes):
+        raise TabularSourceAccessError(f"Missing scope: {required_scope}")
+    return session, principal
+
+
+def _json_row(row: Mapping[str, object]) -> dict[str, Any]:
+    normalized = {str(key).strip(): value for key, value in row.items()}
+    if not normalized or any(not key for key in normalized):
+        raise TabularSourceValidationError("Every snapshot row needs named columns.")
+    try:
+        json.dumps(normalized, default=_unsupported_json)
+    except (TypeError, ValueError) as exc:
+        raise TabularSourceValidationError(f"Snapshot values must be JSON compatible: {exc}") from exc
+    return json.loads(json.dumps(normalized, default=_unsupported_json))
+
+
+def _unsupported_json(value: object) -> object:
+    if isinstance(value, (datetime, Decimal)):
+        return str(value)
+    raise TypeError(f"{type(value).__name__} is not JSON serializable")
+
+
+def _type_name(value: object) -> str:
+    if isinstance(value, bool):
+        return "boolean"
+    if isinstance(value, int):
+        return "integer"
+    if isinstance(value, (float, Decimal)):
+        return "number"
+    if isinstance(value, str):
+        return "string"
+    if isinstance(value, list):
+        return "array"
+    if isinstance(value, dict):
+        return "object"
+    return type(value).__name__.lower()
+
+
+def _column_payload(column: TabularColumn) -> dict[str, object]:
+    return {
+        "name": column.name,
+        "data_type": column.data_type,
+        "nullable": column.nullable,
+    }
+
+
+def _actor_id(principal: ApiPrincipal) -> str | None:
+    return principal.account_id or principal.membership_id or principal.identity_id
+
+
+def _clean_optional(value: str | None) -> str | None:
+    cleaned = str(value or "").strip()
+    return cleaned or None
+
+
+def _escape_like(value: str) -> str:
+    return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+
+
+__all__ = [
+    "ADMIN_SCOPE",
+    "MAX_READ_ROWS",
+    "MAX_SNAPSHOT_BYTES",
+    "MAX_SNAPSHOT_ROWS",
+    "READ_SCOPE",
+    "SqlTabularSourceProvider",
+    "WRITE_SCOPE",
+    "infer_schema",
+    "parse_csv_snapshot",
+    "snapshot_fingerprint",
+]

src/govoplan_connectors/py.typed

@@ -0,0 +1 @@
+

tests/test_datasource_origins.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.datasources import DatasourceOriginReadRequest
+from govoplan_core.core.tabular_sources import TabularSnapshotInput
+from govoplan_core.db.base import Base
+from govoplan_connectors.backend.datasource_origins import (
+    ConnectorDatasourceOriginProvider,
+)
+from govoplan_connectors.backend.db.models import ConnectorTabularSource
+from govoplan_connectors.backend.tabular_sources import (
+    WRITE_SCOPE,
+    SqlTabularSourceProvider,
+)
+
+
+def principal(*, scopes: tuple[str, ...]) -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id="account-1",
+            membership_id="membership-1",
+            tenant_id="tenant-1",
+            scopes=frozenset(scopes),
+        ),
+        account=object(),
+        user=object(),
+    )
+
+
+class ConnectorDatasourceOriginTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(
+            self.engine,
+            tables=[ConnectorTabularSource.__table__],
+        )
+        self.Session = sessionmaker(bind=self.engine)
+        self.session = self.Session()
+        provider = SqlTabularSourceProvider()
+        self.source = provider.create_snapshot(
+            self.session,
+            principal(scopes=(WRITE_SCOPE,)),
+            snapshot=TabularSnapshotInput(
+                name="Imported cases",
+                source_name="imported_cases",
+                rows=({"id": 1, "name": "Ada"},),
+            ),
+        )
+        self.session.commit()
+        self.origins = ConnectorDatasourceOriginProvider(provider)
+
+    def tearDown(self) -> None:
+        self.session.close()
+        Base.metadata.drop_all(
+            self.engine,
+            tables=[ConnectorTabularSource.__table__],
+        )
+        self.engine.dispose()
+
+    def test_datasource_reader_can_discover_and_read_connector_origin(self) -> None:
+        datasource_principal = principal(
+            scopes=("datasources:catalogue:read",),
+        )
+
+        origins = self.origins.list_origins(
+            self.session,
+            datasource_principal,
+        )
+        result = self.origins.read_origin(
+            self.session,
+            datasource_principal,
+            request=DatasourceOriginReadRequest(origin_ref=self.source.ref),
+        )
+
+        self.assertEqual((self.source.ref,), tuple(item.ref for item in origins))
+        self.assertEqual(("live", "cached"), origins[0].supported_modes)
+        self.assertEqual(({"id": 1, "name": "Ada"},), result.rows)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_manifest.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+import unittest
+
+from govoplan_core.core.tabular_sources import (
+    CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER,
+    CAPABILITY_CONNECTORS_TABULAR_SOURCES,
+)
+from govoplan_connectors.backend.manifest import manifest
+
+
+class ConnectorsManifestTests(unittest.TestCase):
+    def test_manifest_exposes_versioned_tabular_capabilities(self) -> None:
+        self.assertEqual("connectors", manifest.id)
+        self.assertIn("access", manifest.optional_dependencies)
+        self.assertIn(
+            "connectors.tabular_sources",
+            {interface.name for interface in manifest.provides_interfaces},
+        )
+        self.assertIn(
+            CAPABILITY_CONNECTORS_TABULAR_SOURCES,
+            manifest.capability_factories,
+        )
+        self.assertIn(
+            CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER,
+            manifest.capability_factories,
+        )
+        self.assertIsNotNone(manifest.migration_spec)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_migrations.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from alembic.runtime.migration import MigrationContext
+from sqlalchemy import create_engine, inspect
+
+from govoplan_connectors.backend.manifest import get_manifest
+from govoplan_core.db.migrations import migrate_database
+
+
+class ConnectorsMigrationTests(unittest.TestCase):
+    def test_baseline_creates_connector_tables_and_head(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="govoplan-connectors-migration-") as directory:
+            url = f"sqlite:///{Path(directory) / 'connectors.db'}"
+            migrate_database(
+                database_url=url,
+                enabled_modules=("connectors",),
+                manifest_factories=(get_manifest,),
+            )
+            engine = create_engine(url)
+            try:
+                with engine.connect() as connection:
+                    self.assertIn(
+                        "e6b7c8d9f0a1",
+                        set(MigrationContext.configure(connection).get_current_heads()),
+                    )
+                    self.assertIn(
+                        "connector_tabular_sources",
+                        inspect(connection).get_table_names(),
+                    )
+            finally:
+                engine.dispose()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_tabular_sources.py

@@ -0,0 +1,179 @@
+from __future__ import annotations
+
+import unittest
+
+from fastapi import HTTPException
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.tabular_sources import (
+    TabularReadRequest,
+    TabularSnapshotInput,
+    TabularSourceAccessError,
+    TabularSourceValidationError,
+)
+from govoplan_core.db.base import Base
+from govoplan_connectors.backend.db.models import ConnectorTabularSource
+from govoplan_connectors.backend.router import api_create_tabular_snapshot
+from govoplan_connectors.backend.schemas import SnapshotCreateRequest
+from govoplan_connectors.backend.tabular_sources import (
+    READ_SCOPE,
+    WRITE_SCOPE,
+    SqlTabularSourceProvider,
+    parse_csv_snapshot,
+)
+
+
+def principal(
+    tenant_id: str = "tenant-1",
+    *,
+    scopes: tuple[str, ...] = (READ_SCOPE, WRITE_SCOPE),
+) -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id="account-1",
+            membership_id="membership-1",
+            tenant_id=tenant_id,
+            scopes=frozenset(scopes),
+        ),
+        account=object(),
+        user=object(),
+    )
+
+
+class ConnectorsTabularSourceTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(self.engine, tables=[ConnectorTabularSource.__table__])
+        self.Session = sessionmaker(bind=self.engine)
+        self.session = self.Session()
+        self.provider = SqlTabularSourceProvider()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        Base.metadata.drop_all(self.engine, tables=[ConnectorTabularSource.__table__])
+        self.engine.dispose()
+
+    def test_snapshot_round_trip_preserves_schema_fingerprint_and_bounds(self) -> None:
+        created = self.provider.create_snapshot(
+            self.session,
+            principal(),
+            snapshot=TabularSnapshotInput(
+                name="Monthly cases",
+                source_name="monthly_cases_2026_07",
+                rows=(
+                    {"case_id": "A-1", "amount": 12, "active": True},
+                    {"case_id": "A-2", "amount": None, "active": False},
+                ),
+            ),
+        )
+        self.session.commit()
+
+        listed = self.provider.list_sources(self.session, principal())
+        preview = self.provider.read_source(
+            self.session,
+            principal(),
+            request=TabularReadRequest(
+                source_ref=created.ref,
+                limit=1,
+                expected_fingerprint=created.fingerprint,
+            ),
+        )
+
+        self.assertEqual((created.ref,), tuple(source.ref for source in listed))
+        self.assertEqual(
+            ["case_id", "amount", "active"],
+            [column.name for column in created.schema],
+        )
+        self.assertEqual(2, preview.total_rows)
+        self.assertEqual(1, len(preview.rows))
+        self.assertTrue(preview.truncated)
+        self.assertEqual(created.fingerprint, preview.source.fingerprint)
+
+    def test_tenant_and_scope_isolation_are_enforced(self) -> None:
+        created = self.provider.create_snapshot(
+            self.session,
+            principal(),
+            snapshot=TabularSnapshotInput(
+                name="Private",
+                source_name="private_source",
+                rows=({"id": 1},),
+            ),
+        )
+        self.session.commit()
+
+        self.assertEqual((), self.provider.list_sources(self.session, principal("tenant-2")))
+        self.assertIsNone(
+            self.provider.get_source(
+                self.session,
+                principal("tenant-2"),
+                source_ref=created.ref,
+            )
+        )
+        with self.assertRaises(TabularSourceAccessError):
+            self.provider.list_sources(
+                self.session,
+                principal(scopes=()),
+            )
+
+    def test_duplicate_source_name_and_stale_fingerprint_are_rejected(self) -> None:
+        snapshot = TabularSnapshotInput(
+            name="Cases",
+            source_name="cases",
+            rows=({"id": 1},),
+        )
+        created = self.provider.create_snapshot(self.session, principal(), snapshot=snapshot)
+        self.session.commit()
+
+        with self.assertRaises(TabularSourceValidationError):
+            self.provider.create_snapshot(self.session, principal(), snapshot=snapshot)
+        with self.assertRaises(TabularSourceValidationError):
+            self.provider.read_source(
+                self.session,
+                principal(),
+                request=TabularReadRequest(
+                    source_ref=created.ref,
+                    expected_fingerprint="stale",
+                ),
+            )
+
+    def test_csv_parser_infers_scalar_values_and_rejects_duplicate_headers(self) -> None:
+        rows = parse_csv_snapshot(
+            "\ufeffid;amount;active;note\n0012;12.5;true;\n2;7;false;ok\n",
+            delimiter=";",
+        )
+
+        self.assertEqual(
+            (
+                {"id": "0012", "amount": 12.5, "active": True, "note": None},
+                {"id": 2, "amount": 7, "active": False, "note": "ok"},
+            ),
+            rows,
+        )
+        with self.assertRaises(TabularSourceValidationError):
+            parse_csv_snapshot("id,id\n1,2\n", delimiter=",")
+        with self.assertRaises(TabularSourceValidationError):
+            parse_csv_snapshot("id,name\n1,Ada,extra\n", delimiter=",")
+
+    def test_malformed_csv_api_request_is_reported_as_validation_error(self) -> None:
+        payload = SnapshotCreateRequest(
+            name="Malformed",
+            source_name="malformed",
+            format="csv",
+            csv_text="id,name\n1,Ada,extra\n",
+        )
+
+        with self.assertRaises(HTTPException) as raised:
+            api_create_tabular_snapshot(
+                payload,
+                session=self.session,
+                principal=principal(),
+            )
+
+        self.assertEqual(422, raised.exception.status_code)
+
+
+if __name__ == "__main__":
+    unittest.main()