receive

Intent

Execute receive to ingest incoming data from external sources and parse it into a structured, validated form for downstream processing.

Success criteria:

• Input data is successfully ingested from the source
• Data is parsed according to expected format
• Validation against schema (if one applies) passes or errors are captured
• Structured output is ready for downstream capabilities
• Malformed or unexpected data is handled gracefully

Compatible schemas:

• schemas/output_schema.yaml
• reference/event_schema.yaml

Inputs

Inputs not listed here are outside the capability's contract: a workflow step that binds them gets W604 (STANDARD v1.1 §5.5).

In a workflow, a step can also wait for a matching event with a required timeout (Standard v1.1 wait@1); see ORCHESTRATION_EXTENSIONS.md.

Procedure

1. Identify input source: Determine where data is coming from
   • File path: Read from local filesystem
   • URL: Fetch from network endpoint
   • Inline: Extract from request payload
   • Stream: Connect to data stream
2. Validate source accessibility: Verify source can be read
   • Check file exists and is readable
   • Verify URL is reachable
   • Confirm stream is connected
   • Check size does not exceed the size limit
3. Detect format: Identify data format if not specified
   • Check content-type headers or file extension
   • Inspect data prefix for format signatures
   • Fall back to text if ambiguous
4. Parse input: Transform raw data into structured form
   • JSON: Parse to object/array using safe parser
   • XML: Parse to DOM or object representation
   • YAML: Parse to object using safe loader
   • CSV: Parse to array of records
   • Text: Split into lines/tokens as appropriate
   • Binary: Extract structured fields per schema
5. Validate structure: Check parsed data against schema
   • If a schema applies to the channel (e.g. schemas/event_v2.yaml): validate and collect errors
   • If no schema: perform basic sanity checks
   • Flag missing required fields
   • Flag type mismatches
6. Extract events/messages: Identify discrete units in the data
   • Single message: wrap as single-element array
   • Batch: extract individual events
   • Stream chunk: identify complete messages
7. Ground evidence: Record provenance for ingested data
   • Source URI or path
   • Timestamp of receipt
   • Size and checksum if applicable
8. Format output: Structure results according to output contract

Output Contract

Return a structured object:

source: string  # Event source: where data came from
timestamp: string  # ISO 8601 date-time of receipt
received:
  source_type: file | url | inline | stream
  format_detected: json | xml | yaml | csv | text | binary
  size_bytes: integer  # Size of received data
  encoding: string  # Character encoding used
data:
  - id: string  # Unique message identifier
    type: string | null  # Event/message type if identifiable
    payload: object  # Parsed message content
    timestamp: string | null  # Message timestamp if present
    metadata: object | null  # Additional message metadata
    sender: string  # Host-authenticated principal that sent the item (with accept_from)
    sender_identity: object  # Sender's identity as the host reported it (identity@1)
parsed_count: integer  # Number of messages/events parsed
validation:
  schema_ref: string | null  # Schema used for validation
  valid: boolean  # Whether all messages passed validation
  errors:
    - message_id: string
      field: string
      error: string
      severity: error | warning
conflicts:
  - type: string  # Type of conflict/anomaly
    description: string
    affected_messages: array[string]
confidence: number  # 0.0-1.0 based on parse success and validation
evidence_anchors: array[string]  # Source references
assumptions: array[string]  # Explicit assumptions

Field Definitions

Examples

Example 1: Receiving JSON API Response

Input:

channel: "https://api.example.com/events"

Output:

source: "https://api.example.com/events"
timestamp: "2024-01-16T12:00:00Z"
received:
  source_type: url
  format_detected: json
  size_bytes: 4256
  encoding: "utf-8"
data:
  - id: "evt_001"
    type: "order.created"
    payload:
      order_id: "ORD-12345"
      customer_id: "CUST-789"
      total: 129.99
      currency: "USD"
    timestamp: "2024-01-16T11:58:30Z"
    metadata:
      source_system: "ecommerce"
      version: "2.1"
  - id: "evt_002"
    type: "order.paid"
    payload:
      order_id: "ORD-12345"
      payment_method: "credit_card"
      transaction_id: "TXN-ABC123"
    timestamp: "2024-01-16T11:59:00Z"
    metadata:
      source_system: "payments"
      version: "1.0"
parsed_count: 2
validation:
  schema_ref: "schemas/event_v2.yaml"
  valid: true
  errors: []
conflicts: []
confidence: 1.0
evidence_anchors:
  - "url:https://api.example.com/events"
  - "schema:schemas/event_v2.yaml"
assumptions:
  - "API response is authoritative source of truth"
  - "Timestamps are in UTC"

Evidence pattern: Source URL recorded; schema validation confirmed; each message given unique ID.

Example 2: Receiving Malformed Data

Input:

channel: "/var/log/sensor_data.json"

Output:

source: "/var/log/sensor_data.json"
timestamp: "2024-01-16T14:30:00Z"
received:
  source_type: file
  format_detected: json
  size_bytes: 1024
  encoding: "utf-8"
data:
  - id: "msg_001"
    type: "sensor_reading"
    payload:
      sensor_id: "SENS-001"
      temperature: 23.5
      humidity: null
      timestamp: "2024-01-16T14:29:00Z"
    timestamp: "2024-01-16T14:29:00Z"
    metadata: null
  - id: "msg_002"
    type: "sensor_reading"
    payload:
      sensor_id: "SENS-002"
      temperature: "ERROR"
      humidity: 65.2
      timestamp: null
    timestamp: null
    metadata: null
parsed_count: 2
validation:
  schema_ref: "schemas/sensor_reading.yaml"
  valid: false
  errors:
    - message_id: "msg_001"
      field: "humidity"
      error: "Required field is null"
      severity: warning
    - message_id: "msg_002"
      field: "temperature"
      error: "Expected number, got string 'ERROR'"
      severity: error
    - message_id: "msg_002"
      field: "timestamp"
      error: "Required field is null"
      severity: error
conflicts:
  - type: "data_quality"
    description: "Sensor SENS-002 reporting error state instead of reading"
    affected_messages: ["msg_002"]
confidence: 0.5
evidence_anchors:
  - "file:/var/log/sensor_data.json"
  - "schema:schemas/sensor_reading.yaml"
assumptions:
  - "File contains most recent sensor batch"
  - "'ERROR' string indicates sensor malfunction"
next_actions:
  - "Investigate sensor SENS-002 for malfunction"
  - "Check why SENS-001 humidity is null"

Example 3: Receiving Webhook Payload

Input:

channel: inline  # payload delivered with the request

Payload received:

<?xml version="1.0"?>
<notification type="alert">
  <severity>high</severity>
  <message>CPU usage exceeded threshold</message>
  <host>server-01</host>
</notification>

Output:

source: "inline"
timestamp: "2024-01-16T15:00:00Z"
received:
  source_type: inline
  format_detected: xml
  size_bytes: 198
  encoding: "utf-8"
data:
  - id: "msg_001"
    type: "notification.alert"
    payload:
      severity: "high"
      message: "CPU usage exceeded threshold"
      host: "server-01"
    timestamp: null
    metadata:
      xml_root: "notification"
      xml_attributes:
        type: "alert"
parsed_count: 1
validation:
  schema_ref: null
  valid: true
  errors: []
conflicts: []
confidence: 0.95
evidence_anchors:
  - "inline:xml:notification"
assumptions:
  - "XML is well-formed and complete"
  - "No schema validation requested"

Verification

• Source is accessible and data retrieved successfully
• Format detection matches actual data format
• All messages parsed without critical errors
• Validation errors logged with actionable detail
• confidence reflects parse success rate

Verification tools: Read (to access file sources), Grep (to search for patterns in text data)

Safety Constraints

• mutation: false
• requires_checkpoint: false
• requires_approval: false
• risk: low

Capability-specific rules:

• Do not execute or run arbitrary code from received data
• Use safe parsing methods (JSON.parse, safe YAML loader)
• Validate size limits before loading large files into memory
• Sanitize file paths to prevent directory traversal
• Do not store credentials or secrets found in payloads
• Log but do not propagate malformed/suspicious data without flagging

Composition Patterns

Commonly follows:

• External triggers (webhooks, API calls, file drops)
• schedule - Scheduled data ingestion

Commonly precedes:

• transform - Normalize received data for processing
• validate - Deeper validation beyond schema
• integrate - Merge received data with existing state
• synchronize - Combine with other data sources
• recall - Check received data against prior context

Anti-patterns:

• Never process received data without validation
• Never trust received data for authentication decisions without verification
• Avoid processing unbounded streams without size limits

Workflow references:

• See reference/composition_patterns.md#digital-twin-sync-loop for receive as entry point
• First step in most data ingestion pipelines