Skill v1.0.1
currentAutomated scan100/100+4 new
name: integrate description: Combine heterogeneous data sources into a unified model with conflict resolution, schema alignment, and provenance tracking. Use when merging data from multiple systems, consolidating information, or building comprehensive views. argument-hint: "[sources] [strategy] [conflict_resolution]" disable-model-invocation: false user-invocable: true allowed-tools: Read, Grep context: fork agent: explore metadata: version: "1.0.1"
Intent
Perform data integration to combine information from multiple heterogeneous sources into a unified, coherent model. This handles schema differences, resolves conflicts, and preserves provenance.
Success criteria:
- All sources are incorporated or explicitly excluded
- Schema mismatches are resolved consistently
- Conflicts are documented with resolution rationale
- Provenance traces every value to its source
- Information loss is minimized and documented
World Modeling Context: Integrate is essential for building comprehensive world-state from multiple observation sources. It combines outputs from retrieve, search, and inspect into unified models, and depends on identity-resolution for entity deduplication.
Hard dependencies:
- Requires
identity-resolutionwhen sources contain overlapping entities
Compatible schemas:
schemas/output_schema.yaml
Inputs
| Parameter | Required | Type | Description | |
|---|---|---|---|---|
sources | Yes | array | Data sources to integrate (files, API outputs, retrieved data) | |
strategy | No | string | Merge strategy | |
conflict_resolution | No | string | How to handle conflicts |
Inputs not listed here are outside the capability's contract: a workflow step that binds them gets W604 (STANDARD v1.1 §5.5).
Procedure
1) Analyze sources: Understand what each source contains
- Inventory entities and attributes per source
- Identify source schemas (explicit or inferred)
- Note data freshness and authority
- Detect overlapping vs. unique content
2) Align schemas: Map source schemas to target
- Identify equivalent fields (same meaning, different names)
- Map types (handle type conversions)
- Handle missing fields (default values, nulls)
- Document unmappable fields (information loss)
3) Resolve entity identity: Determine which records refer to same entity
- Apply
identity-resolutionfor overlapping entities - Assign canonical IDs
- Track all source IDs as aliases
4) Merge attributes: For each entity, combine attributes from sources
- Non-overlapping: Simply include from source
- Overlapping, matching: Confirm consistency
- Overlapping, conflicting: Apply conflict strategy
5) Apply conflict strategy (conflict_resolution):
For prefer_recent:
- Use the most recently updated value
- Requires reliable timestamps
For prefer_authoritative:
- Rank sources by authority
- Use value from most authoritative source
For merge:
- Combine values where possible (lists, sets)
- For scalars, may require manual resolution
For manual:
- Flag all conflicts for human review
- Do not auto-resolve
6) Validate integrated data: Check against target schema
- Required fields present
- Type constraints satisfied
- Relationship integrity maintained
- Custom invariants hold
7) Document integration: Record what happened
- Per-field provenance (which source)
- Conflict resolutions with rationale
- Information loss (unmapped fields)
- Transformation applied
8) Produce output: Generate integrated model
- Conforms to target schema
- Includes provenance metadata
- Documents confidence levels
Output Contract
Return a structured object:
merged:schema: string | object # Target schema reference or inlineentities:- id: string # Canonical IDtype: string # Entity typeattributes: object # Integrated attributessource_map: # Where each attribute came fromattribute_name:source: stringoriginal_value: any | null # If transformedconfidence: numberrelationships:- source: stringtarget: stringtype: stringprovenance: stringobservations: array # Merged observations, when the sources are observation streamsmeta:world_id: string # World model identifier, when integrating into a world modelintegration_report:sources_used: array[string]sources_excluded: array[string]entity_count:total: integerfrom_single_source: integermerged: integerfield_mapping:- source_field: stringtarget_field: stringtransformation: string | nullinformation_loss:- source: stringfield: stringreason: string # unmappable, filtered, etc.impact: low | medium | highvalidation_results:passed: booleanerrors: array[string]warnings: array[string]conflicts:- entity_id: stringfield: stringvalues:- source: stringvalue: anyresolution: stringresolved_value: anyconfidence: numberprovenance:sources: array[object] # Source metadataintegration_timestamp: stringconfidence: numberconfidence: number # 0.0-1.0 overallevidence_anchors: array[string]assumptions: array[string]
Field Definitions
| Field | Type | Description | |
|---|---|---|---|
merged | object | The unified data model (entities, relationships; observations and meta.world_id when merging observation streams into a world model) | |
merged.entities[].source_map | object | Per-attribute source tracking | |
integration_report | object | Sources used/excluded, entity counts, field mapping, information loss, validation | |
integration_report.information_loss | array | Data that couldn't be integrated | |
conflicts | array | All conflicts found and how each was resolved | |
provenance | object | Source metadata and integration timestamp | |
confidence | number | 0.0-1.0 overall integration confidence | |
evidence_anchors | array | References to each source consulted | |
assumptions | array | Assumptions made while integrating |
Examples
Example 1: Integrating Service Data from Multiple Sources
Input:
sources:- name: "kubernetes"type: "api"data:- id: "api-gateway-deployment"status: "Running"replicas: 3last_updated: "2025-01-24T10:00:00Z"- name: "datadog"type: "api"data:- service_name: "api-gateway"health: "healthy"latency_p99: 45last_updated: "2025-01-24T10:05:00Z"- name: "package.json"type: "file"data:- name: "api-gateway"version: "2.3.1"dependencies: ["express", "cors"]conflict_resolution: prefer_recent
Target schema: microservices_v1.
Output:
merged:schema: "microservices_v1"entities:- id: "api-gateway"type: "Service"attributes:name: "api-gateway"version: "2.3.1"status: "healthy"replicas: 3latency_p99: 45dependencies: ["express", "cors"]source_map:name:source: "package.json"original_value: nullconfidence: 1.0version:source: "package.json"original_value: nullconfidence: 1.0status:source: "datadog"original_value: "healthy"confidence: 0.9replicas:source: "kubernetes"original_value: nullconfidence: 1.0latency_p99:source: "datadog"original_value: nullconfidence: 0.95dependencies:source: "package.json"original_value: nullconfidence: 1.0relationships: []integration_report:sources_used: ["kubernetes", "datadog", "package.json"]sources_excluded: []entity_count:total: 1from_single_source: 0merged: 1field_mapping:- source_field: "kubernetes.status"target_field: "status"transformation: "Mapped 'Running' to 'healthy'"- source_field: "datadog.service_name"target_field: "name"transformation: nullinformation_loss:- source: "kubernetes"field: "labels"reason: "No target schema field for Kubernetes labels"impact: lowvalidation_results:passed: trueerrors: []warnings:- "Service has no APIs defined"conflicts:- entity_id: "api-gateway"field: "status"values:- source: "kubernetes"value: "Running"- source: "datadog"value: "healthy"resolution: "prefer_recent (datadog more recent)"resolved_value: "healthy"confidence: 0.85provenance:sources:- name: "kubernetes"authority: highfreshness: "2025-01-24T10:00:00Z"- name: "datadog"authority: mediumfreshness: "2025-01-24T10:05:00Z"- name: "package.json"authority: highfreshness: "2025-01-20T00:00:00Z"integration_timestamp: "2025-01-24T12:00:00Z"confidence: 0.9confidence: 0.9evidence_anchors:- "kubernetes_api:deployments/api-gateway-deployment"- "datadog:services/api-gateway"- "package.json:1-20"assumptions:- "All sources refer to the same service"- "Datadog health status more accurate than K8s Running state"
Evidence pattern: Mapped fields from three sources to unified schema, resolved status conflict using recency.
Example 2: Integrating Documentation with Code Analysis
Input:
sources:- name: "code_analysis"data:functions:- name: "processOrder"file: "src/orders.ts"parameters: ["orderId", "options"]- name: "api_docs"data:endpoints:- path: "/orders/process"method: "POST"description: "Process a pending order"parameters:- name: "order_id"required: trueconflict_resolution: merge
Target schema: api_documentation.
Output:
merged:schema: "api_documentation"entities:- id: "endpoint_process_order"type: "Endpoint"attributes:path: "/orders/process"method: "POST"description: "Process a pending order"implementation: "src/orders.ts:processOrder"parameters:- name: "order_id"code_name: "orderId"required: truesource: "body"source_map:path: { source: "api_docs" }method: { source: "api_docs" }description: { source: "api_docs" }implementation: { source: "code_analysis" }parameters: { source: "merged" }integration_report:sources_used: ["code_analysis", "api_docs"]entity_count:total: 1merged: 1information_loss: []conflicts:- entity_id: "endpoint_process_order"field: "parameters[0].name"values:- source: "code_analysis"value: "orderId"- source: "api_docs"value: "order_id"resolution: "merge - both names preserved (code vs API)"resolved_value: { name: "order_id", code_name: "orderId" }confidence: 0.95confidence: 0.95evidence_anchors:- "src/orders.ts:processOrder"- "api_docs:endpoints/process"
Verification
- [ ] All source entities are accounted for (integrated or documented as excluded)
- [ ] Conflicts have documented resolutions
- [ ] Output validates against target schema
- [ ] Provenance traces every attribute to source
- [ ] Information loss is documented
Verification tools: Schema validators, provenance checkers
Safety Constraints
mutation: falserequires_checkpoint: falserequires_approval: falserisk: low
Capability-specific rules:
- Never silently drop data - document all information loss
- When confidence < 0.5 on a merge, flag for review
- Preserve source identifiers for traceability
- Apply
identity-resolutionbefore merging overlapping entities
Composition Patterns
Commonly follows:
retrieve- Get data from sources before integratingsearch- Discover relevant sources to integrateidentity-resolution- Resolve entity identity before merge
Commonly precedes:
world-state- Integration produces unified stategrounding- Integrated data needs groundingdiff-world-state- Compare integrated states over time
Anti-patterns:
- Never integrate without understanding source schemas
- Avoid ignoring conflicts - always resolve or flag
Workflow references:
- See
reference/composition_patterns.md#world-model-buildfor integration in model construction - See
reference/composition_patterns.md#digital-twin-sync-loopfor ongoing integration