2026-07-22-agentplat-generic-ingestion-design.md

agentplat generic ingestion — design

Date: 2026-07-22 Repos: mozilla/agentplat (app), mozilla/webservices-infra (chart: mzcld-demo/k8s/agentplat)

Problem

The agentplat chart currently deploys one hand-written workload per integration: slack-listener (Deployment), jira-poller (CronJob), and upstream now also metric-poller (CronJob). Each new integration means a new bespoke workload in Helm. We want the chart to be integration-agnostic: the set of k8s workloads should not grow or change when integrations are added.

Key insight from the code

The control plane is already the generic router. Every ingestion entrypoint is a near-identical shell that fetches raw events from a source and POSTs to /api/runs {deployment, input, source_event_id}. The app already has a pluggable connector registry (@registry.connector("jira-new-ticket"), "raw event → agent input"). What it lacks is the mirror seam for the outbound fetch ("how do I pull raw events from source X"); that logic is copy-pasted in each *_poller.py / *_listener.py main(). _post_run duplication is already flagged as tech debt in the app's STATUS.md.

Additionally, each AgentDeployment (*.deploy.yaml) already declares its triggers: [{type, config, input_mapping}], and the control plane loads all deployments. So the app already holds the full list of "which triggers are active and how they're configured" — the chart never needs to know.

Design

App side (mozilla/agentplat)

1. Source registry — mirror the connector registry:

   SOURCES: dict[str, Source] = {}
   
   @registry.source("jira-new-ticket", mode="poll")
   def jira_source(config, creds):        # yields (raw_event, source_event_id)
       ...
   
   @registry.source("slack-app-mention", mode="listen")
   def slack_source(config, creds, emit): # blocks; calls emit(raw, source_event_id)
       ...

   Each source declares a mode: poll (fetch periodically) or listen (hold a persistent subscription). mode is the only irreducible axis — Slack Socket Mode must be listen.

2. Generic ingest entrypoint — python -m agentplat.ingest listen|poll:

   • On start, query the control plane for active deployments/triggers, filter to the sources whose mode matches the subcommand, and run them.
   • listen: start every matched source concurrently (each on its own task/thread) and block forever.
   • poll: run an internal scheduler that polls each matched source at its own interval (from the deployment's trigger config). Long-running (see chart).
   • For each event, POST the raw event to the control plane tagged with the trigger type, letting the control plane normalize in-plane (the same path /webhooks/alert already uses). Do NOT pre-normalize / ride the cli passthrough — the app maintainers already consider that the weaker path.
   • Single shared post_run client (removes the duplicated _post_run).
   • slack_listener.py, jira_poller.py, metric_poller.py collapse into ingest.py + three small @source functions under providers/.

3. Control-plane trigger discovery — a small read endpoint (e.g. GET /api/triggers) that returns active triggers grouped by mode, derived from the already-loaded deployments. The ingest runners consume this.

4. Trigger-typed ingest — /api/runs (or a new /ingest) must accept an explicit trigger.type + raw event and normalize via the connector registry. This also makes future push webhooks (/webhooks/jira, etc.) trivial.

Chart side (mzcld-demo/k8s/agentplat)

Two generic, integration-agnostic long-running Deployments plus the control plane. Neither mentions Slack/Jira/metric:

• agentplat-ingest-listeners — python -m agentplat.ingest listen
• agentplat-ingest-pollers — python -m agentplat.ingest poll

Both get AGENTPLAT_CONTROL_URL and envFrom agentplat-secrets. The pollers Deployment uses an internal scheduler, so all schedules live app-side (in the deployment trigger config) — the chart carries no schedule and no source list.

Adding an integration = a new @source function in the app + an AgentDeployment that references its trigger. The chart is never touched.

What stays infra-side (unavoidable)

Credentials. Both runner pods mount the tenant secret (envFrom agentplat-secrets) so sources can read their creds. Which sources consume which keys is app-side; the chart just mounts the secret and stays agnostic.

Decisions (agreed)

• Pollers: long-running Deployment with an internal per-source scheduler (not a CronJob), so schedules are app-side.
• Discovery: runners auto-discover active triggers from the control plane (no source list in Helm or runner env).
• Normalize in-plane (post raw + trigger type), not in the runner.

Convergence toward webhooks (future, not required)

Once ingest is trigger-typed, push sources become HTTP routes on the control plane (grafana-alert already is). Jira/metric support push and could drop the poller Deployment entirely, exposed via an HTTPRoute. End state: control plane (all webhooks) + one listeners Deployment + zero pollers. This design converges there without a rewrite.

Sequencing

1. App: source registry + ingest.py + GET /api/triggers + trigger-typed ingest; port the three existing entrypoints to @source functions.
2. Chart: replace the three per-integration workloads with the two generic ingest Deployments.

Open questions

• Exact shape of the control-plane trigger-discovery response and how per-source interval is expressed in *.deploy.yaml.
• How the pollers Deployment handles a source with no interval (default? skip?).
• Whether the listeners Deployment should crash/restart vs. degrade if one source fails to connect (one bad source shouldn't take down the others).