from __future__ import annotations import asyncio import inspect import time from dataclasses import dataclass, field from typing import Any, Callable, Optional from core.content_pipeline import ContentGate from core.content.dedup import DedupStore from core.infra.audit import AuditLog from core.infra.metrics import Metrics from core.jobs.queue import ProcessingQueue from core.pipeline.adapter import SectionPipelineAdapter from core.providers.pool import AIProviderPool from core.ai.gateway import AIGateway from core.publishing.ledger import PublishLedger from core.publishing.publisher import IdempotentPublisher from core.health.monitor import HealthMonitor from core.repair.engine import AutoRepairEngine from core.repair.policy import RepairGuard, RepairPolicy from core.repair.guarded import GuardedRepair from core.sections.base import SectionConfig from core.sections.registry import SectionRegistry from core.runtime.state_store import RuntimeStateStore from core.control.plane import ControlPlane from core.security.secret_registry import SecretRegistry from core.security.hardening import SecurityHardener @dataclass class RuntimeResult: status: str stage: str reason: str = "" data: Any = None fingerprint: str = "" job_id: str = "" @dataclass class SectionRuntimeState: enabled: bool = True blocked_words: list[str] = field(default_factory=list) duplicate_protection: bool = True ai_enabled: bool = True auto_repair_enabled: bool = True sources: list[str] = field(default_factory=list) class RuntimeIntegration: """The phase-15 composition root. Every section uses the same ordered runtime: gate -> normalization -> durable queue -> AI gateway -> validation -> idempotent publisher -> health/audit/metrics. External Telegram/Blogger implementations are injected as adapters so the core remains testable and future sections do not require core edits. """ def __init__( self, db: Any = None, *, db_path: str = "data/p29_runtime.sqlite3", project_root: str = ".", blocked_words: Optional[list[str]] = None, ai_gateway: Optional[AIGateway] = None, queue: Optional[ProcessingQueue] = None, publisher: Optional[IdempotentPublisher] = None, health: Optional[HealthMonitor] = None, metrics: Optional[Metrics] = None, audit: Optional[AuditLog] = None, auto_repair: Optional[GuardedRepair] = None, repair_proposer: Optional[Callable[[str, dict], Optional[str]]] = None, ): self.db = db self.db_path = db_path self.state_store = RuntimeStateStore(db_path) self.queue = queue or ProcessingQueue(db_path) self.pool = ai_gateway.pool if ai_gateway else AIProviderPool() self.ai = ai_gateway or AIGateway(self.pool) self.ledger = publisher.ledger if publisher else PublishLedger(db_path) self.publisher = publisher or IdempotentPublisher(self.ledger, {}) self.health = health or HealthMonitor() self.metrics = metrics or Metrics() self.audit = audit or AuditLog("data/audit.jsonl") self.secret_registry = SecretRegistry() self.security = SecurityHardener(self.secret_registry) self.security.refresh() if auto_repair is None: policy = RepairPolicy( enabled=True, auto_apply_low_risk=False, require_admin_for_medium=True, require_admin_for_high=True, ) engine = AutoRepairEngine(project_root) guard = RepairGuard(policy) auto_repair = GuardedRepair(engine, guard) self.auto_repair = auto_repair self.repair_proposer = repair_proposer self.health.on_unhealthy = self._on_unhealthy self._gates: dict[str, ContentGate] = {} self._sections: dict[str, SectionRuntimeState] = { "blogger": SectionRuntimeState(), "news": SectionRuntimeState(), "sports": SectionRuntimeState(), } self._load_section_state() self._publisher_adapters: dict[str, Callable] = {} self._section_pipelines: dict[str, SectionPipelineAdapter] = {} self._repair_enabled = bool(self.state_store.get("control:auto_repair", True)) self.control_plane = ControlPlane(self) self._register_health() def _load_section_state(self) -> None: for section, default in list(self._sections.items()): raw = self.state_store.get(f"section:{section}", {}) or {} if not isinstance(raw, dict): continue for key in ("enabled", "blocked_words", "duplicate_protection", "ai_enabled", "auto_repair_enabled", "sources"): if key in raw: setattr(default, key, raw[key]) def _persist_section_state(self, section: str) -> None: state = self._sections[section] self.state_store.set(f"section:{section}", { "enabled": bool(state.enabled), "blocked_words": list(state.blocked_words), "duplicate_protection": bool(state.duplicate_protection), "ai_enabled": bool(state.ai_enabled), "auto_repair_enabled": bool(state.auto_repair_enabled), "sources": list(state.sources), }) def _register_health(self) -> None: for section in self._sections: self.health.register( f"section:{section}", lambda section=section: self._section_health(section), ) for provider in self.pool.providers(): self.health.register( f"provider:{provider}", lambda provider=provider: self._provider_health(provider), ) def _section_health(self, section: str) -> dict: state = self._sections[section] return { "enabled": state.enabled, "queue": self.queue.store.get_stats(), "ai_enabled": state.ai_enabled, } def _provider_health(self, provider: str) -> dict: snap = self.pool.snapshot().get(provider, []) if not snap: raise RuntimeError("no provider keys available") available = [x for x in snap if x["cooldown"] <= 0] if not available: raise RuntimeError("all provider keys are in cooldown") return {"keys": len(snap), "available_keys": len(available)} def register_ai_provider(self, provider: str, adapter: Callable) -> None: self.ai.register(provider, adapter) self.health.register( f"provider:{provider.lower()}", lambda provider=provider.lower(): self._provider_health(provider), ) def register_publisher(self, target: str, adapter: Callable) -> None: self._publisher_adapters[target] = adapter self.publisher.adapters[target] = adapter def set_section_state(self, section: str, **changes) -> SectionRuntimeState: state = self._sections.setdefault(section, SectionRuntimeState()) for key, value in changes.items(): if not hasattr(state, key): raise KeyError(key) setattr(state, key, value) self._gates.pop(section, None) self._section_pipelines.pop(section, None) self._persist_section_state(section) self._audit("section.state_changed", section=section, changes=changes) return state def get_section_state(self, section: str) -> dict: state = self._sections.setdefault(section, SectionRuntimeState()) return { "section": section, "enabled": state.enabled, "blocked_words": list(state.blocked_words), "duplicate_protection": state.duplicate_protection, "ai_enabled": state.ai_enabled, "auto_repair_enabled": state.auto_repair_enabled, "sources": list(state.sources), } def _gate(self, section: str) -> ContentGate: if section not in self._gates: words = list(self._sections.setdefault(section, SectionRuntimeState()).blocked_words) state = self._sections.setdefault(section, SectionRuntimeState()) self._gates[section] = ContentGate( _OverlayGateDB(self.db, words, section), dedup=DedupStore(self.db_path) if state.duplicate_protection else None, ) return self._gates[section] def ingest( self, section: str, article: str, article_id: str, *, source: str = "", source_url: str = "", channel_id: str = "", target: str = "", metadata: Optional[dict] = None, ) -> RuntimeResult: state = self._sections.setdefault(section, SectionRuntimeState()) if not state.enabled: self.metrics.inc(f"{section}.rejected.disabled") return RuntimeResult("rejected", "section", "section_disabled") if state.sources and source and source not in state.sources: self.metrics.inc(f"{section}.rejected.source") self._audit("content.rejected", section=section, article_id=article_id, reason="source_not_allowed") return RuntimeResult("rejected", "source", "source_not_allowed") if state.sources and not source: self.metrics.inc(f"{section}.rejected.source") return RuntimeResult("rejected", "source", "source_required") gate = self._gate(section) # Blocked words are evaluated before fingerprint/dedup and therefore before # queue/AI/API usage. verdict = gate.preflight(article, source_url, channel_id, source=source, article_id=article_id) if not verdict.allowed: self.metrics.inc(f"{section}.rejected.{verdict.reason}") self._audit( "content.rejected", section=section, article_id=article_id, reason=verdict.reason, matched=list(verdict.matched), ) return RuntimeResult( "rejected", "gate", verdict.reason, fingerprint=verdict.fingerprint, ) cleaned = self._clean_for_queue(article, channel_id) if not cleaned: self.metrics.inc(f"{section}.rejected.empty_after_cleanup") self._audit("content.rejected", section=section, article_id=article_id, reason="empty_after_cleanup") return RuntimeResult("rejected", "cleanup", "empty_after_cleanup", fingerprint=verdict.fingerprint) scoped_id = f"{section}:{article_id}" payload = { "section": section, "article_id": scoped_id, "source": source, "source_url": source_url, "channel_id": channel_id, "target": target or section, "article": cleaned, "fingerprint": verdict.fingerprint, "metadata": metadata or {}, } try: job_id = self.queue.enqueue_article(scoped_id, payload) except Exception: # A failed queue write must not leave a permanently reserved dedup # record in the persistent gate. try: gate.dedup.forget(verdict.fingerprint) except Exception: pass self.metrics.inc(f"{section}.queue.error") self._audit("queue.enqueue_failed", section=section, article_id=article_id) raise self.metrics.inc(f"{section}.queued") self._audit( "content.queued", section=section, article_id=article_id, fingerprint=verdict.fingerprint, job_id=job_id, ) return RuntimeResult( "queued", "queue", data=payload, fingerprint=verdict.fingerprint, job_id=job_id, ) def _clean_for_queue(self, text: str, channel_id: str = "") -> str: """Apply deterministic removal terms only after the zero-AI gate.""" terms = [] if self.db is not None: try: terms.extend(self.db.get_global_remove_terms() or []) except Exception: pass if channel_id: try: terms.extend(self.db.get_channel_delete_terms(channel_id) or []) except Exception: pass for raw in terms: term = str(raw or "").strip() if term: text = text.replace(term, "") return text.strip() def process_payload(self, payload: dict) -> RuntimeResult: section = payload["section"] state = self._sections.setdefault(section, SectionRuntimeState()) if not state.enabled: return RuntimeResult("rejected", "section", "section_disabled") if not state.ai_enabled: return RuntimeResult("rejected", "ai", "ai_disabled") text = payload["article"] try: response = self.ai.article_package( text, task=f"process_{section}_article", ) data = response.data if not isinstance(data, dict): raise RuntimeError("AI returned non-object article package") self.metrics.inc(f"{section}.ai.success") self.metrics.inc(f"provider.{response.provider}.success") except Exception as exc: self.metrics.inc(f"{section}.ai.failure") self._audit("ai.failure", section=section, error=str(exc)) raise # The same gate runs after AI so generated content cannot bypass safety. gate = self._gate(section) article = dict(data) article["source_url"] = payload.get("source_url", "") article["fingerprint"] = payload.get("fingerprint", "") post = gate.postflight(article, payload.get("channel_id", "")) if not post.allowed: self.metrics.inc(f"{section}.rejected.postflight") self._audit( "content.postflight_rejected", section=section, reason=post.reason, matched=list(post.matched), ) return RuntimeResult("rejected", "validation", post.reason, fingerprint=post.fingerprint) article["_runtime"] = { "section": section, "article_id": payload["article_id"], "target": payload.get("target") or section, "fingerprint": payload.get("fingerprint", ""), } self.metrics.inc(f"{section}.validated") return RuntimeResult("validated", "validation", data=article, fingerprint=payload.get("fingerprint", "")) def publish_payload(self, payload: dict) -> RuntimeResult: article = payload["article"] meta = article.get("_runtime", {}) target = payload.get("target") or meta.get("target") or payload.get("section") article_id = meta.get("article_id") or payload.get("article_id") content = self._article_content(article) try: result = self.publisher.publish( target, article_id, content, article=article, section=payload.get("section"), ) except Exception as exc: self.metrics.inc(f"{payload.get('section','unknown')}.publish.failure") self._audit("publish.failure", section=payload.get("section"), error=str(exc)) raise self.metrics.inc(f"{payload.get('section','unknown')}.publish.{result.get('status','unknown')}") self._audit( "publish.result", section=payload.get("section"), article_id=article_id, status=result.get("status"), remote_id=result.get("remote_id"), ) return RuntimeResult(result.get("status", "unknown"), "publisher", data=result) @staticmethod def _article_content(article: dict) -> str: # Keep a stable representation for idempotency hashing while allowing # adapters to consume the structured article through the `article=` kwarg. for key in ("body", "article", "content", "text"): if article.get(key): return str(article[key]) return str(article) def handle_job(self, payload: dict, job: Optional[dict] = None) -> RuntimeResult: """Worker handler: queue -> AI -> validation -> publisher. The persistent queue's retry/failure policy handles worker crashes. Publish idempotency is checked inside IdempotentPublisher before the remote call. """ processed = self.process_payload(payload) if processed.status != "validated": return processed payload = dict(payload) payload["article"] = processed.data published = self.publish_payload(payload) return published def build_concurrent_workers(self, workers: int = 2, worker_lease_timeout: float = 300): """Build a bounded concurrent worker pool on the canonical persistent queue.""" from core.jobs.concurrency import ConcurrentWorkers return ConcurrentWorkers(self.queue.store, self.worker_handlers(), workers=workers, lease_timeout=worker_lease_timeout) def load_snapshot(self) -> dict: """Safe operational load snapshot; no article payloads are exposed.""" return { "queue": self.queue.store.get_stats(), "workers": {"supported": True, "max_concurrency": 32}, "metrics": self.metrics.detailed_snapshot(), } def build_worker(self, worker_id: Optional[str] = None): from core.jobs.worker import JobWorker return JobWorker(self.queue.store, self.worker_handlers(), worker_id=worker_id) def worker_handlers(self) -> dict[str, Callable]: return {"article.process": self.handle_job} def build_section_adapter(self, section: str) -> SectionPipelineAdapter: if section not in self._section_pipelines: self._section_pipelines[section] = SectionPipelineAdapter( section, self._gate(section), self.queue, self.ai, self.publisher, ) return self._section_pipelines[section] def health_snapshot(self) -> dict: return self.health.snapshot() def metrics_snapshot(self) -> dict: return self.metrics.snapshot() def control_snapshot(self) -> dict: return { "sections": {k: self.get_section_state(k) for k in self._sections}, "health": self.health_snapshot(), "metrics": self.metrics_snapshot(), "metrics_detailed": self.metrics.detailed_snapshot(), "providers": self.pool.snapshot(), "queue": self.queue.store.get_stats(), "alerts": self.control_plane.evaluate(), "auto_repair": { "enabled": self._repair_enabled, "policy": self.auto_repair.guard.policy.__dict__, }, } def set_auto_repair(self, enabled: bool) -> dict: self._repair_enabled = bool(enabled) self.state_store.set("control:auto_repair", self._repair_enabled) self._audit("repair.toggled", enabled=self._repair_enabled) return {"ok": True, "enabled": self._repair_enabled} def operational_snapshot(self) -> dict: return self.control_plane.snapshot() def recent_audit(self, limit: int = 50) -> list[dict]: return self.audit.recent(limit) def acknowledge_alert(self, key: str, *, admin_approved=False) -> dict: return self.control_plane.acknowledge(key, admin_approved=admin_approved) def queue_snapshot(self, status: str | None = None, limit: int = 50) -> list[dict]: rows = self.queue.store.list_jobs(status=status, limit=limit) # Keep control-plane output safe: do not expose arbitrary payload content. return [{ "id": row.get("id"), "kind": row.get("kind"), "status": row.get("status"), "attempts": row.get("attempts"), "available_at": row.get("available_at"), "claimed_at": row.get("claimed_at"), "worker_id": row.get("worker_id"), "last_error": row.get("last_error", ""), } for row in rows] def requeue_dead_job(self, job_id: str, *, admin_approved: bool = False) -> dict: if not admin_approved: return {"ok": False, "reason": "admin_required"} ok = self.queue.store.requeue_dead(job_id) self._audit("queue.dead_requeued", job_id=job_id, ok=ok) return {"ok": ok, "job_id": job_id} def _on_unhealthy(self, name: str, service) -> None: self.metrics.inc("health.unhealthy") self._audit("health.unhealthy", service=name, failures=service.consecutive_failures) if not self._repair_enabled or not self.repair_proposer: return if name.startswith("section:"): section = name.split(":", 1)[1] if not self._sections.get(section, SectionRuntimeState()).auto_repair_enabled: return try: patch = self.repair_proposer(name, service.__dict__.copy()) if not patch: return record = self.auto_repair.test_and_apply( f"health:{name}", patch, tests=["tests/v29"], admin_approved=False, ) self._audit( "repair.result", service=name, repair_id=record.repair_id, status=record.status, reason=record.reason, ) except Exception as exc: self._audit("repair.failure", service=name, error=str(exc)) def monitor_once(self) -> dict: result = self.health.check_all() for name, service in result.items(): if service.status == "unhealthy": self.metrics.inc("health.unhealthy") return self.health_snapshot() def _audit(self, event: str, **data) -> None: # AuditLog redacts string values. No raw key/token is intentionally passed. self.audit.write(event, **data) class _OverlayGateDB: def __init__(self, legacy, words, section): self.legacy = legacy self.words = list(words) self.section = section def get_blocked_words(self): out = list(self.words) if self.legacy is not None: try: out.extend(self.legacy.get_blocked_words() or []) except Exception: pass return list(dict.fromkeys(out)) def get_channel_blocked_words(self, channel_id): if self.legacy is not None: try: return self.legacy.get_channel_blocked_words(channel_id) or [] except Exception: pass return [] def is_published(self, fp): if self.legacy is not None: try: return bool(self.legacy.is_published(fp)) except Exception: pass return False def get_article(self, fp): if self.legacy is not None: try: return self.legacy.get_article(fp) except Exception: pass return None def get_all_articles(self): if self.legacy is not None: try: return self.legacy.get_all_articles() or [] except Exception: pass return [] class _MemoryGateDB: """Minimal DB contract used by isolated runtime tests.""" def __init__(self, words): self.words = list(words) self.articles = {} self.published = set() def get_blocked_words(self): return self.words def get_channel_blocked_words(self, channel_id): return [] def is_published(self, fp): return fp in self.published def get_article(self, fp): return self.articles.get(fp) def get_all_articles(self): return list(self.articles.values())