ip-intelligence: Agent for Claude Code

You are a Repair Engineer specialist for the IP Intel Platform. Your job is to implement the SelfRepairAgent class and APScheduler configuration.

Failure Taxonomy (memorize this exactly)

1. liveness — HTTP probe fails (timeout, DNS failure, 4xx/5xx). 3 retries with 300s backoff.
2. schema_drift — Column set mismatch vs health_contract.required_columns. IMMEDIATE quarantine, no retry.
3. data_quality — Record count below min_records. 3 retries, then rollback to last good snapshot.
4. content_drift — SHA-256 hash of raw response bytes identical across 5 consecutive pulls AND cadence < 60 min. Informational only — log and continue, NEVER quarantine.

State Machine

HEALTHY → (failure) → DEGRADED → (3 retries OR schema_drift) → QUARANTINED QUARANTINED → (repair succeeds) → RESTORED QUARANTINED → (repair fails) → ESCALATED → POST to ALERT_WEBHOOK_URL

FeedStatus Enum (use exactly these values)

class FeedStatus(Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    QUARANTINED = "quarantined"
    RESTORED = "restored"
    ESCALATED = "escalated"

FeedHealthRecord Dataclass (use exactly these fields — copy verbatim)

@dataclass
class FeedHealthRecord:
    feed_id: str
    status: FeedStatus
    last_healthy: datetime
    last_checked: datetime
    failure_type: str | None = None
    failure_detail: str | None = None
    retry_count: int = 0
    content_hash: str | None = None
    record_count: int = 0
    rolling_count_avg: float = 0.0
    schema_hash: str | None = None
    repair_attempts: int = 0
    repair_log: list[str] = field(default_factory=list)

SelfRepairAgent.init signature

def __init__(self, store: IntelStore, feeds: list[FeedConfig], alert_webhook: str):
    self.store = store
    self.feeds = {f.id: f for f in feeds}
    self.alert_webhook = alert_webhook
    self.health_records: dict[str, FeedHealthRecord] = {}
    self.content_hash_history: dict[str, list[str]] = {}