content-find-agent-new/content_agent/integrations/composite_runtime.py

from __future__ import annotations

from pathlib import Path
from typing import Any

from content_agent.interfaces import RuntimeStore


class CompositeRuntimeStore:
    def __init__(self, primary: RuntimeStore, export: RuntimeStore) -> None:
        self.primary = primary
        self.export = export

    def prepare_run(self, run_id: str) -> Path:
        self.primary.prepare_run(run_id)
        return self.export.prepare_run(run_id)

    def run_dir(self, run_id: str) -> Path:
        return self.export.run_dir(run_id)

    def write_json(self, run_id: str, filename: str, data: dict[str, Any]) -> Path:
        self.primary.write_json(run_id, filename, data)
        return self.export.write_json(run_id, filename, data)

    def update_json(self, run_id: str, filename: str, data: dict[str, Any]) -> Path:
        self.primary.update_json(run_id, filename, data)
        return self.export.update_json(run_id, filename, data)

    def append_jsonl(self, run_id: str, filename: str, rows: list[dict[str, Any]]) -> Path:
        self.primary.append_jsonl(run_id, filename, rows)
        return self.export.append_jsonl(run_id, filename, rows)

    def read_json(self, run_id: str, filename: str) -> dict[str, Any]:
        try:
            return self.primary.read_json(run_id, filename)
        except Exception:
            return self.export.read_json(run_id, filename)

    def read_jsonl(self, run_id: str, filename: str) -> list[dict[str, Any]]:
        try:
            return self.primary.read_jsonl(run_id, filename)
        except Exception:
            return self.export.read_jsonl(run_id, filename)

    def file_status(self, run_id: str) -> dict[str, bool]:
        try:
            return self.primary.file_status(run_id)
        except Exception:
            return self.export.file_status(run_id)

    def create_run_record(self, record: dict[str, Any]) -> None:
        self.primary.create_run_record(record)
        self.export.create_run_record(record)

    def update_run_record(self, run_id: str, updates: dict[str, Any]) -> None:
        self.primary.update_run_record(run_id, updates)
        self.export.update_run_record(run_id, updates)

    def record_policy_run(self, record: dict[str, Any]) -> None:
        self.primary.record_policy_run(record)
        self.export.record_policy_run(record)

    def append_run_event_records(
        self,
        run_id: str,
        policy_run_id: str,
        rows: list[dict[str, Any]],
    ) -> None:
        self.primary.append_run_event_records(run_id, policy_run_id, rows)
        self.export.append_run_event_records(run_id, policy_run_id, rows)

    def write_publish_jobs(
        self,
        run_id: str,
        policy_run_id: str,
        rows: list[dict[str, Any]],
    ) -> None:
        self.primary.write_publish_jobs(run_id, policy_run_id, rows)
        self.export.write_publish_jobs(run_id, policy_run_id, rows)

    def write_author_assets(self, rows: list[dict[str, Any]]) -> None:
        self.primary.write_author_assets(rows)
        self.export.write_author_assets(rows)

    def write_author_asset_roles(self, rows: list[dict[str, Any]]) -> None:
        self.primary.write_author_asset_roles(rows)
        self.export.write_author_asset_roles(rows)

    def write_search_clue_assets(self, rows: list[dict[str, Any]]) -> None:
        self.primary.write_search_clue_assets(rows)
        self.export.write_search_clue_assets(rows)

    def write_search_clue_asset_evidence(self, rows: list[dict[str, Any]]) -> None:
        self.primary.write_search_clue_asset_evidence(rows)
        self.export.write_search_clue_asset_evidence(rows)

    def read_performance_feedback(
        self,
        run_id: str,
        policy_run_id: str,
    ) -> list[dict[str, Any]]:
        return self.primary.read_performance_feedback(run_id, policy_run_id)

    def enqueue_media_pipeline_task(self, task: dict[str, Any]) -> dict[str, Any]:
        primary_method = getattr(self.primary, "enqueue_media_pipeline_task", None)
        if callable(primary_method):
            stored = primary_method(task)
        else:
            stored = task
        export_method = getattr(self.export, "enqueue_media_pipeline_task", None)
        if callable(export_method):
            export_method(task)
        return stored

    def update_media_pipeline_task(self, task_id: str, updates: dict[str, Any]) -> None:
        for store in (self.primary, self.export):
            method = getattr(store, "update_media_pipeline_task", None)
            if callable(method):
                method(task_id, updates)

    def read_media_pipeline_tasks(
        self,
        run_id: str,
        *,
        batch_id: str | None = None,
    ) -> list[dict[str, Any]]:
        method = getattr(self.primary, "read_media_pipeline_tasks", None)
        if callable(method):
            return method(run_id, batch_id=batch_id)
        method = getattr(self.export, "read_media_pipeline_tasks", None)
        if callable(method):
            return method(run_id, batch_id=batch_id)
        return []