AI Source Connector Health Check: Stop Agents From Trusting Broken Data

An AI agent can sound confident while reading yesterday's permissions, a half-synced CRM record, or a document connector that silently stopped crawling. That is worse than a normal outage because the UI still works, the model still answers, and users may not notice the data is wrong until trust is already damaged.

If you are building AI features on top of customer data, your connectors are now part of the answer quality system. A Slack import, Google Drive sync, database replica, support-ticket feed, analytics warehouse, or MCP data tool is not just plumbing. It is the evidence layer your agent uses to decide what is true.

This guide shows how to build an AI source connector health check: a practical set of tests, scores, alerts, and fallback rules that stop agents from trusting broken data.

No product pitch here. The pattern works whether you use managed connectors, open-source ingestion, custom sync jobs, MCP tools, RAG pipelines, or direct database access.

Why connector health suddenly matters more

Recent AI platform activity points in the same direction: more agentic workflows, more data-source connectors, more plugin-style integrations, and more pressure to measure cost per task. Product launches around AI sources and agent plugins show that builders want agents to work across real systems, not toy prompts. Developer discussions around federated query layers, integrations, MCP, and agent databases show the same demand from the technical side.

That creates a quiet failure mode.

Traditional software usually fails loudly when a dependency breaks. A 500 error, empty response, expired token, or failed cron job is visible. AI systems can fail softly. They retrieve fewer documents, use stale facts, skip restricted rows, quote an old policy, or answer from cached context.

The model may still produce a polished response.

For AI app builders, connector health affects:

• answer accuracy
• tenant isolation
• retrieval quality
• cost per task
• support escalations
• user trust
• compliance evidence
• agent action safety

If an agent drafts emails, updates tickets, analyzes revenue, answers policy questions, or triggers workflow actions, bad source data is not a minor bug. It becomes bad judgment at machine speed.

The core idea: every source needs a health contract

A connector health check is not one ping endpoint. It is a contract that says, "This source is fresh, complete enough, permission-safe, schema-compatible, and usable for this AI task."

A useful health contract has five layers:

1. Connection health: Can we reach the source and authenticate?
2. Sync health: Are records arriving on time and without large gaps?
3. Schema health: Do fields still match what retrieval, prompts, and tools expect?
4. Permission health: Are tenant, user, and role filters still enforced?
5. Answer health: Can the AI workflow answer known questions using this source?

Most teams monitor the first layer. Production AI features need all five.

A practical source health score

Use a score that is simple enough for alerts and strict enough to protect users. Here is a starting model:

Check	Weight	Failure example
Auth and API reachability	15%	expired OAuth token
Freshness lag	20%	latest synced ticket is 9 hours old
Sync completeness	15%	import skipped 18% of documents
Schema compatibility	15%	customer_id renamed to account_id
Permission filters	20%	user can retrieve another tenant's row
Sample answer tests	15%	agent cannot answer a known policy question

A source below 90 is usable with caution. Below 80 should degrade the AI feature. Below 70 should block high-risk answers or actions.

The exact numbers matter less than the behavior: the agent should know when evidence is unhealthy.

Health check data model

Start with a small table. You can expand later.

CREATE TABLE ai_source_health_checks (
  id UUID PRIMARY KEY,
  tenant_id UUID NOT NULL,
  source_id TEXT NOT NULL,
  source_type TEXT NOT NULL,
  checked_at TIMESTAMPTZ NOT NULL,
  status TEXT NOT NULL, -- healthy, degraded, blocked
  score INT NOT NULL,
  freshness_lag_seconds INT,
  schema_version TEXT,
  permission_test_passed BOOLEAN,
  sample_answer_passed BOOLEAN,
  failure_reasons JSONB NOT NULL DEFAULT '[]',
  evidence JSONB NOT NULL DEFAULT '{}'
);

CREATE INDEX idx_ai_source_health_latest
ON ai_source_health_checks (tenant_id, source_id, checked_at DESC);

The important part is evidence. Do not store only a green or red status. Store what was checked, which sample records were used, what changed, and which workflow should degrade.

Check 1: freshness lag

Freshness is the easiest failure to miss. A connector can look healthy while serving old data.

Track the source's latest update time, the latest synced record time, and the latest indexed or embedded time. Those are different clocks.

type FreshnessResult = {
  sourceId: string;
  latestSourceUpdate: string;
  latestSyncedRecord: string;
  latestIndexedRecord: string;
  lagSeconds: number;
  status: "healthy" | "degraded" | "blocked";
};

function scoreFreshness(lagSeconds: number, maxLagSeconds: number) {
  if (lagSeconds <= maxLagSeconds) return { score: 100, status: "healthy" };
  if (lagSeconds <= maxLagSeconds * 3) return { score: 70, status: "degraded" };
  return { score: 0, status: "blocked" };
}

Set freshness targets by workflow, not globally.

• Support responses may need ticket data within minutes.
• Contract search may tolerate a few hours.
• Quarterly analytics summaries may tolerate a daily warehouse sync.
• Agent actions against production records should require current permissions.

A single freshness threshold creates false confidence.

Check 2: sync completeness

Fresh data is not enough if the connector skipped half the source.

Measure expected versus observed records:

• records seen in the source API
• records accepted by ingestion
• records rejected by validation
• records indexed for retrieval
• records removed due to permissions
• records too large or malformed to process

A practical completeness check can be simple:

function completenessRatio(stats: {
  expected: number;
  ingested: number;
  indexed: number;
}) {
  if (stats.expected === 0) return 1;
  return Math.min(stats.ingested, stats.indexed) / stats.expected;
}

function scoreCompleteness(ratio: number) {
  if (ratio >= 0.98) return 100;
  if (ratio >= 0.90) return 70;
  return 20;
}

Also track the reason records were skipped. "Ten documents failed because they were corrupt" is different from "all private documents disappeared after a permission bug."

Check 3: schema compatibility

AI features often depend on fields that are not obvious in the UI: owner IDs, timestamps, product area, customer tier, source URL, embedding text, permission tags, or lifecycle status.

If a connector changes a field name, enum value, null behavior, or nested JSON shape, the model may still receive text, but the workflow logic can break.

Create a schema manifest per source:

{
  "source_type": "support_tickets",
  "schema_version": "tickets.v3",
  "required_fields": [
    "ticket_id",
    "tenant_id",
    "requester_id",
    "status",
    "updated_at",
    "body_text",
    "permission_scope"
  ],
  "enum_fields": {
    "status": ["open", "pending", "solved", "closed"]
  }
}

Then validate samples on every sync and before major agent runs. If a required field disappears, do not let the agent guess.

Check 4: permission probes

Permission bugs are the most dangerous connector failures because retrieval can look accurate while leaking the wrong data.

Run permission probes for each tenant and role pattern:

• A user should retrieve their own documents.
• A user should not retrieve another tenant's documents.
• A restricted user should not retrieve admin-only records.
• A revoked user should retrieve nothing after revocation.
• A service agent should only retrieve the scopes granted to that workflow.

Example probe:

type PermissionProbe = {
  actorId: string;
  tenantId: string;
  query: string;
  mustInclude?: string[];
  mustExclude: string[];
};

async function runPermissionProbe(probe: PermissionProbe, retrieve: Function) {
  const results = await retrieve({
    tenantId: probe.tenantId,
    actorId: probe.actorId,
    query: probe.query,
    limit: 20
  });

  const ids = results.map((r: any) => r.recordId);
  const leaked = probe.mustExclude.filter(id => ids.includes(id));

  return {
    passed: leaked.length === 0,
    leaked,
    resultCount: results.length
  };
}

Run these probes against the same retrieval path your AI feature uses. Testing only the database policy is not enough if embeddings, caches, search indexes, or tool responses bypass it.

Check 5: sample answer tests

Connector health should end with a question: can the AI workflow still answer known tasks from this source?

Build a tiny golden set per source:

Source	Test question	Expected evidence
Docs	"What is the refund window for annual plans?"	policy page URL + current section
Tickets	"What are the top three billing complaints this week?"	recent tagged tickets
CRM	"Which renewal accounts are blocked by security review?"	account records with status
Analytics	"Did activation improve after onboarding change?"	metric definition + date range

The model does not need to match exact wording. It does need to retrieve the right evidence and avoid unsupported claims.

A simple judge rubric:

{
  "retrieved_required_evidence": true,
  "used_current_records": true,
  "respected_permissions": true,
  "answer_contains_unsupported_claims": false,
  "status": "pass"
}

This is where connector health joins RAG evaluation. Retrieval metrics tell you whether the system found relevant chunks. Source health tells you whether those chunks should be trusted in the first place.

How agents should use health status

Do not hide source health inside dashboards only. Pass a compact health summary into the agent workflow.

{
  "source_health": {
    "support_tickets": {
      "status": "degraded",
      "score": 76,
      "reason": "freshness lag is 4h 12m; target is 30m",
      "allowed_actions": ["draft", "summarize_with_warning"],
      "blocked_actions": ["send_reply", "update_ticket_status"]
    }
  }
}

This lets the agent adapt:

• answer with a freshness warning
• ask for confirmation before acting
• use a fallback source
• switch to draft-only mode
• refuse high-risk actions
• create an internal incident note

The key rule: unhealthy evidence should reduce autonomy.

Degraded UX beats silent confidence

A good degraded state is honest and useful. Avoid vague banners like "Something went wrong." Tell the user what is safe.

Better examples:

• "I can draft a reply, but I will not send it because ticket data is 4 hours out of date."
• "Analytics are available through yesterday. I cannot answer questions about today's usage yet."
• "This answer excludes private Drive documents because the permission sync is being repaired."
• "I found matching records, but source health is degraded, so please review before applying changes."

Users forgive temporary limits. They do not forgive confident wrong answers.

Alerting that avoids noise

Do not page someone every time a connector has a small delay. Alert by risk and user impact.

Useful alert dimensions:

• affected tenants
• affected workflows
• source type
• action risk level
• freshness lag
• failed permission probes
• sample answer failures
• number of AI runs that used degraded data

A high-risk alert should say:

Connector health blocked: support_tickets
Tenant: acme
Reason: permission probe failed
Impact: send_reply and update_ticket_status disabled
Recent agent runs using this source: 12
Next step: rotate connector token, rebuild permission index, replay probes

That is much better than "sync failed."

Where this fits in your architecture

A connector health service usually sits between ingestion and AI runtime.

Source API / DB / File Store
        ↓
Connector Sync Job
        ↓
Validation + Permission Index + Embeddings
        ↓
Source Health Checks
        ↓
AI Runtime / Agent / RAG / MCP Tool
        ↓
Answer Receipt + Logs

For small teams, this can be a scheduled job and one database table. You do not need a separate platform on day one.

Start with:

1. freshness lag
2. schema checks
3. permission probes
4. five sample-answer tests
5. runtime degradation rules

Then add completeness scoring, trend reports, and per-workflow thresholds.

Common mistakes

• Treating OAuth success as health. A valid token only proves access, not freshness, completeness, permissions, or answer quality.
• Testing ingestion but not retrieval. Test the path the agent actually uses.
• Using one global status. A source can be healthy for summaries and unsafe for actions.
• Ignoring deletes and revocations. Old data must disappear from search, caches, memory, and tool results.
• Letting the model decide trust from text alone. Enforce hard runtime policy outside the model.

Implementation checklist

Use this as a first sprint plan:

• [ ] List every source your AI feature can read.
• [ ] Define freshness targets per workflow.
• [ ] Store latest source, sync, and index timestamps.
• [ ] Add schema manifests for required fields.
• [ ] Track expected, ingested, rejected, and indexed records.
• [ ] Create permission probes for normal, restricted, revoked, and cross-tenant users.
• [ ] Build five sample-answer tests per critical source.
• [ ] Store health scores with evidence, not just status.
• [ ] Pass compact source health into the AI runtime.
• [ ] Block or degrade high-risk actions when source health is low.
• [ ] Show honest user-facing degraded states.
• [ ] Attach source health to answer receipts and incident reviews.

FAQ

What is an AI source connector health check?

It is a set of tests that verifies whether a data source is reachable, fresh, complete, schema-compatible, permission-safe, and usable by an AI workflow. It goes beyond checking whether the API is online.

Is this different from RAG evaluation?

Yes. RAG evaluation checks whether retrieval and answers are good. Source connector health checks whether the underlying data source should be trusted before retrieval or agent action uses it. They work best together.

Do small AI products need connector health checks?

Yes, but they can start small. Track freshness, schema compatibility, permission probes, and a few sample-answer tests. That is enough to catch many silent failures before users do.

How often should connector health checks run?

Run lightweight checks after every sync and before high-risk agent workflows. Run deeper sample-answer tests on a schedule, after schema changes, and after permission or ingestion code changes.

Should agents see source health status?

Yes, but do not rely on the model alone. Pass a compact health summary to the agent for better responses, and enforce hard blocks in runtime policy for risky actions.

What should happen when a connector is unhealthy?

The workflow should degrade based on risk. Low-risk summaries can show warnings. Drafting can continue with review. Writes, sends, billing actions, and cross-user updates should pause until health is restored.

Final thought

AI quality is not only a model problem. It is an evidence problem.

If your agent reads broken data, stale permissions, or incomplete syncs, a better prompt will only make the wrong answer sound cleaner. Build connector health checks early, wire them into runtime behavior, and make unhealthy evidence impossible to ignore.