from dataclasses import dataclass, field
from typing import Any, Optional
from datetime import datetime

@dataclass
class TaskHeader:
    pipeline_id: str
    quality_floor: float = 0.75
    metadata: dict = field(default_factory=dict)

@dataclass
class CanonicalPayload:
    source_text: str
    extracted_items: list[dict]
    context_window: Optional[str] = None
    version: str = "1.0"

@dataclass
class PipelineIR:
    task_header: TaskHeader
    payload: CanonicalPayload
    provenance_log: list[dict] = field(default_factory=list)

class ObligationClassifierAdapter:
    def __init__(self, model_interface: Any):
        self._model = model_interface

    def ingress(self, ir: PipelineIR) -> list[dict]:
        """Transform canonical IR into classifier-native input."""
        formatted_inputs = []
        for item in ir.payload.extracted_items:
            formatted_inputs.append({
                "entity_category": item.get("label

", "UNKNOWN"), "reference_text": ir.payload.source_text[:120], "confidence_threshold": ir.task_header.quality_floor, "processing_flags": ir.task_header.metadata.get("flags", {}) }) return formatted_inputs

def egress(self, raw_output: list[dict], ir: PipelineIR) -> PipelineIR:
    """Transform classifier output back into canonical IR."""
    for result in raw_output:
        ir.payload.extracted_items.append({
            "entity_type": result.get("entity_category"),
            "obligation_role": result.get("classification"),
            "score": result.get("confidence", 0.0)
        })
    return ir

The ingress function extracts the `label` field from the canonical payload and maps it to `entity_category`, which matches the classifier's expected schema. It also pulls the context window directly from the source text and applies the pipeline's quality threshold. All translation happens here. Downstream models receive standardized IR objects. Upstream models never see the classifier's internal schema.

### Step 3: Route Through the Canonical IR

The orchestrator no longer handles translation. It simply passes the IR between adapters.

```python
def execute_pipeline(ir: PipelineIR, adapters: list) -> PipelineIR:
    for adapter in adapters:
        native_input = adapter.ingress(ir)
        raw_output = adapter._model.predict(native_input)
        ir = adapter.egress(raw_output, ir)
        
        # Append provenance entry
        ir.provenance_log.append({
            "adapter_class": adapter.__class__.__name__,
            "timestamp": datetime.utcnow().isoformat(),
            "confidence_avg": sum(
                item.get("score", 0.0) 
                for item in ir.payload.extracted_items
            ) / max(len(ir.payload.extracted_items), 1),
            "status": "completed"
        })
    return ir

The orchestrator remains schema-agnostic. It only manages execution order and provenance collection. This separation of concerns prevents pipeline logic from leaking into model boundaries.

Step 4: Enforce Immutable Provenance

The provenance chain is append-only by design. Each adapter execution records the model identifier, execution timestamp, average confidence score, and status. Because entries are never modified or reordered, the chain serves as a tamper-evident audit trail. In production environments processing regulated data, this chain satisfies compliance requirements without additional logging infrastructure.

Architecture Rationale

• Adapters own translation: Placing ingress/egress logic inside the adapter ensures schema changes are localized. The model team maintains their interface contract.
• Canonical IR as routing contract: Standardizing data flow eliminates N*(N-1)/2 connectors. Every model speaks the same language.
• Thresholds via task headers: Pipeline-level configuration (quality floors, feature flags) flows through the header, preventing hardcoding inside adapters.
• Append-only provenance: Immutable logs guarantee audit integrity. Debugging becomes deterministic because execution history cannot be altered.

Pitfall Guide

1. Embedding Translation in the Orchestrator

Explanation: Developers often place field mapping logic inside the pipeline runner to avoid creating adapter classes. This couples the orchestrator to every model's schema. Fix: Move all translation into ingress/egress methods. The orchestrator should only handle IR routing and provenance collection.

2. Ignoring Schema Versioning in the IR

Explanation: Treating the canonical IR as static leads to silent data loss when models introduce new fields or deprecate old ones. Fix: Include a version field in the payload. Implement backward-compatible adapters that map legacy fields to current IR structures. Use semantic versioning for IR releases.

3. Hardcoding Business Logic in Adapters

Explanation: Adapters sometimes contain validation rules, threshold calculations, or domain-specific filtering. This violates the single-responsibility principle and makes adapters difficult to test. Fix: Keep adapters strictly focused on schema translation. Route business rules through the task header or a dedicated validation middleware that operates on the canonical IR.

4. Breaking Provenance Immutability

Explanation: Teams occasionally overwrite provenance entries to correct confidence scores or update status flags. This destroys audit integrity. Fix: Treat the provenance log as append-only. If corrections are needed, append a new entry with a correction flag and reference the original index. Never mutate existing records.

5. Over-Fetching Context Windows

Explanation: Ingress adapters frequently slice large text blocks to satisfy model context limits. Blindly truncating text drops critical semantic boundaries. Fix: Implement sentence-aware chunking. Pass the context_window through the IR payload so downstream adapters can request specific segments without re-parsing raw text.

6. Skipping Adapter Validation

Explanation: Deploying adapters without schema validation causes runtime failures when upstream models change output structures. Fix: Run the adapter against fixture datasets before deployment. Use the SDK's validation command to verify conformance against expected input/output contracts.

7. Treating Adapters as Stateless Utilities

Explanation: Assuming adapters are pure functions ignores the need for model-specific initialization, caching, or connection pooling. Fix: Design adapters as stateful objects that manage their own dependencies. Initialize model clients, load configuration, and handle retries inside the adapter class, not in global pipeline state.

Production Bundle

Action Checklist

• Define canonical IR schema with explicit versioning and extensible payload structure
• Implement ingress adapters that map IR fields to model-native inputs without business logic
• Implement egress adapters that normalize model outputs back to canonical format
• Route all pipeline execution through the IR; remove point-to-point connector functions
• Attach append-only provenance entries after each adapter execution
• Validate adapters against fixture datasets before registering with the pipeline registry
• Monitor adapter latency and translation error rates in production observability dashboards

Decision Matrix

Scenario	Recommended Approach	Why	Cost Impact
Rapid prototyping with fixed models	Point-to-point connectors	Faster initial setup; schema changes are rare	Low upfront, high long-term maintenance
Multi-team model deployment with independent release cycles	Canonical IR + Adapters	Isolates schema changes; enables parallel development	Moderate upfront, near-zero marginal cost per update
Regulated data processing (HIPAA/GDPR)	Canonical IR + Immutable Provenance	Provides tamper-evident audit trail without custom logging	Higher storage cost, significant compliance savings
High-throughput inference with strict latency budgets	Adapter-level batching + IR routing	Reduces network hops; keeps translation close to model	Increased memory usage, improved throughput

Configuration Template

# adapter_config.yaml
adapter:
  class: ObligationClassifierAdapter
  model_endpoint: "https://inference.internal/classify/v2"
  timeout_ms: 1200
  retry_policy:
    max_attempts: 2
    backoff_multiplier: 1.5

ingress:
  field_mappings:
    ir_label: "entity_category"
    ir_context_slice: 120
  threshold_source: "task_header.quality_floor"

egress:
  output_normalization:
    confidence_field: "score"
    role_field: "obligation_role"
  provenance:
    enabled: true
    capture_latency: true
    capture_cost: true

Quick Start Guide

1. Install the SDK: Run pip install synapse-adapter-sdk to access the canonical IR utilities and validation tooling.
2. Define your IR contract: Create a dataclass or schema definition that standardizes payload structure, task headers, and provenance logging.
3. Build your first adapter: Implement ingress and egress methods that translate between the canonical IR and your model's native format. Keep business logic out of these methods.
4. Validate locally: Execute synapse-validate --adapter your_module.YourAdapter --all-fixtures to verify schema conformance before pipeline integration.
5. Wire the orchestrator: Pass the IR through your adapter chain. The orchestrator should only handle execution order and provenance collection. Deploy and monitor translation latency.