"""G6 원천 원장에서 교수자 attention/gap/Phase 3 산출물을 직접 생산한다.

HTTP 호출자가 이미 계산한 신호를 주입하는 기존 control-plane과 달리 이 모듈은
append-only G0~G5 원장을 읽고, 재현 가능한 파생 산출물만 G6 저장소에 기록한다.
원문 발화는 읽거나 복제하지 않는다.
"""

from __future__ import annotations

import asyncio
import hashlib
import json
import logging
from collections import defaultdict
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from typing import Any
from uuid import UUID, uuid5

import asyncpg

from .. import db
from ..config import settings
from ..contracts.supervision_research import (
    LedgerEvidencePointer,
    LearnerAttentionSignal,
    Phase3EvidenceArtifact,
)
from . import supervision_research_store, supervision_research_version_evaluator


_PRODUCER_NAMESPACE = UUID("f28dd79f-cda7-43c2-8b41-93e7ba7f4de7")
_PRODUCER_TASK: asyncio.Task[None] | None = None
logger = logging.getLogger(__name__)
_UNRESOLVED_RUPTURE_STATES = {
    "onset",
    "recognized",
    "repair_attempted",
    "missed",
    "partial",
}


@dataclass(frozen=True, slots=True)
class DerivedSignal:
    learner_id: UUID
    competency_id: str
    signal: LearnerAttentionSignal


@dataclass(frozen=True, slots=True)
class ManifestInput:
    artifacts: tuple[Phase3EvidenceArtifact, ...]
    source_by_domain: Mapping[str, tuple[UUID, LedgerEvidencePointer]]
    source_fingerprint: str


def _value(row: Mapping[str, Any], key: str, default: Any = None) -> Any:
    value = row.get(key, default)
    return default if value is None else value


def _canonical_hash(payload: Any) -> str:
    canonical = json.dumps(
        payload,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
        default=str,
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def _learner_ref(learner_id: UUID) -> str:
    return f"learner-{learner_id.hex[:20]}"


def _signal_id(*parts: object) -> str:
    digest = _canonical_hash([str(value) for value in parts])[:24]
    return f"oas-g6-signal-{digest}"


def _route(session_id: object) -> str:
    return f"/teacher/sessions/{session_id}"


def _pointer(
    *,
    ledger: str,
    event_id: object,
    session_id: object | None,
) -> LedgerEvidencePointer:
    return LedgerEvidencePointer(
        ledger=ledger,
        event_id=str(event_id),
        session_id=str(session_id) if session_id else None,
        route_hint=_route(session_id) if session_id else "/teacher/dashboard",
    )


def _latest_session_assessment(
    assessment: Mapping[str, Any],
) -> Mapping[str, Any] | None:
    sessions = assessment.get("sessions")
    if not isinstance(sessions, list):
        return None
    candidates = [item for item in sessions if isinstance(item, dict)]
    if not candidates:
        return None
    return max(candidates, key=lambda item: int(item.get("session_no") or 0))


def _trajectory_uncertainty(session: Mapping[str, Any]) -> float:
    axes = session.get("axes")
    if not isinstance(axes, list):
        return 1.0
    values = [
        float(item["uncertainty"])
        for item in axes
        if isinstance(item, dict) and item.get("uncertainty") is not None
    ]
    return max(values) if values else 1.0


def _interval_width(payload: Mapping[str, Any]) -> float:
    interval = payload.get("error_interval") or payload.get("success_interval")
    if not isinstance(interval, dict):
        return 1.0
    try:
        return max(0.0, min(1.0, float(interval["upper"]) - float(interval["lower"])))
    except (KeyError, TypeError, ValueError):
        return 1.0


def derive_attention_signals(
    *,
    trajectory_rows: Sequence[Mapping[str, Any]],
    rupture_rows: Sequence[Mapping[str, Any]],
    calibration_rows: Sequence[Mapping[str, Any]],
    transfer_rows: Sequence[Mapping[str, Any]],
    practice_rows: Sequence[Mapping[str, Any]],
    safety_rows: Sequence[Mapping[str, Any]] = (),
) -> tuple[DerivedSignal, ...]:
    """파생 규칙은 관찰된 원장 상태만 사용하며 임상 진단을 만들지 않는다."""

    output: list[DerivedSignal] = []
    for row in safety_rows:
        # 원천 테이블은 nullable 참조와 JSON detail을 허용한다. 유효한 회기에
        # 커밋된 escalation만 채택하고 detail·trigger 원문·PII는 투영하지 않는다.
        if row.get("escalated") is not True:
            continue
        event_id = row.get("safety_event_id")
        trigger_type = row.get("trigger_type")
        session_value = row.get("session_id")
        learner_value = row.get("learner_id")
        sequence_value = row.get("observed_sequence")
        if (
            not isinstance(event_id, int)
            or isinstance(event_id, bool)
            or event_id < 1
            or not isinstance(trigger_type, str)
            or not trigger_type.strip()
            or session_value is None
            or learner_value is None
            or not isinstance(sequence_value, int)
            or isinstance(sequence_value, bool)
            or sequence_value < 1
        ):
            continue
        try:
            learner_id = UUID(str(learner_value))
            session_id = UUID(str(session_value))
        except (TypeError, ValueError):
            continue
        output.append(
            DerivedSignal(
                learner_id=learner_id,
                competency_id="competency.safety_boundary",
                signal=LearnerAttentionSignal(
                    signal_id=_signal_id("safety", event_id),
                    learner_ref=_learner_ref(learner_id),
                    signal_type="safety_boundary",
                    severity="high",
                    state="active",
                    # app.safety_events에는 보정된 uncertainty가 없으므로
                    # 확신도를 만들어내지 않고 unknown(1.0)을 보존한다.
                    uncertainty=1.0,
                    observed_sequence=sequence_value,
                    evidence=(
                        _pointer(
                            ledger="safety_event",
                            event_id=event_id,
                            session_id=session_id,
                        ),
                    ),
                ),
            )
        )

    for row in trajectory_rows:
        assessment = _value(row, "assessment", {})
        if not isinstance(assessment, dict):
            continue
        latest = _latest_session_assessment(assessment)
        if latest is None:
            continue
        trajectory_status = str(latest.get("status") or "")
        if trajectory_status not in {"off_track", "deteriorating"}:
            continue
        learner_id = UUID(str(row["learner_id"]))
        revision_id = row["revision_id"]
        pointer = _pointer(
            ledger="outcome_trajectory_revision",
            event_id=revision_id,
            session_id=row.get("anchor_session_id"),
        )
        output.append(
            DerivedSignal(
                learner_id=learner_id,
                competency_id="competency.outcome_monitoring",
                signal=LearnerAttentionSignal(
                    signal_id=_signal_id("trajectory", revision_id, trajectory_status),
                    learner_ref=_learner_ref(learner_id),
                    signal_type="deterioration",
                    severity="high"
                    if trajectory_status == "deteriorating"
                    else "moderate",
                    state="active",
                    uncertainty=_trajectory_uncertainty(latest),
                    observed_sequence=max(1, int(_value(row, "revision_no", 1))),
                    evidence=(pointer,),
                ),
            )
        )

    for row in rupture_rows:
        rupture_state = str(_value(row, "to_state", ""))
        if rupture_state not in _UNRESOLVED_RUPTURE_STATES:
            continue
        learner_id = UUID(str(row["learner_id"]))
        observation_id = row["observation_id"]
        output.append(
            DerivedSignal(
                learner_id=learner_id,
                competency_id="competency.rupture_repair",
                signal=LearnerAttentionSignal(
                    signal_id=_signal_id("rupture", observation_id, rupture_state),
                    learner_ref=_learner_ref(learner_id),
                    signal_type="unresolved_rupture",
                    severity="high" if rupture_state == "missed" else "moderate",
                    state="active",
                    uncertainty=float(_value(row, "uncertainty", 1.0)),
                    observed_sequence=max(1, int(_value(row, "sequence_no", 1))),
                    evidence=(
                        _pointer(
                            ledger="rupture_observation_event",
                            event_id=observation_id,
                            session_id=row.get("session_id"),
                        ),
                    ),
                ),
            )
        )

    calibration_groups: dict[tuple[UUID, str], list[Mapping[str, Any]]] = defaultdict(
        list
    )
    for row in calibration_rows:
        key = (UUID(str(row["learner_id"])), str(row["competency_id"]))
        calibration_groups[key].append(row)
    for (learner_id, competency_id), rows in calibration_groups.items():
        ordered = sorted(
            rows, key=lambda item: int(_value(item, "snapshot_no", 0)), reverse=True
        )
        recent = ordered[:2]
        if len(recent) < 2:
            continue
        payloads = [_value(item, "assessment_payload", {}) for item in recent]
        if not all(
            isinstance(item, dict) and item.get("bias") == "overconfident"
            for item in payloads
        ):
            continue
        latest = recent[0]
        latest_payload = payloads[0]
        snapshot_id = latest["assessment_snapshot_id"]
        output.append(
            DerivedSignal(
                learner_id=learner_id,
                competency_id=competency_id,
                signal=LearnerAttentionSignal(
                    signal_id=_signal_id("calibration", snapshot_id, "overconfident"),
                    learner_ref=_learner_ref(learner_id),
                    signal_type="persistent_overconfidence",
                    severity="moderate",
                    state="active",
                    uncertainty=_interval_width(latest_payload),
                    observed_sequence=max(1, int(_value(latest, "snapshot_no", 1))),
                    evidence=(
                        _pointer(
                            ledger="calibration_assessment",
                            event_id=snapshot_id,
                            session_id=latest.get("session_id"),
                        ),
                    ),
                    counterevidence=tuple(
                        str(value)
                        for value in latest_payload.get("counterevidence", [])
                    ),
                ),
            )
        )

    for row in transfer_rows:
        payload = _value(row, "assessment_payload", {})
        if (
            not isinstance(payload, dict)
            or payload.get("transfer_verified") is not False
        ):
            continue
        learner_id = UUID(str(row["learner_id"]))
        assessment_id = row["transfer_assessment_id"]
        eligible = bool(payload.get("eligible"))
        output.append(
            DerivedSignal(
                learner_id=learner_id,
                competency_id=str(row["competency_id"]),
                signal=LearnerAttentionSignal(
                    signal_id=_signal_id("transfer", assessment_id, "failed"),
                    learner_ref=_learner_ref(learner_id),
                    signal_type="transfer_failure",
                    severity="moderate" if eligible else "low",
                    state="active" if eligible else "monitoring",
                    uncertainty=_interval_width(payload),
                    observed_sequence=max(1, int(_value(row, "observed_sequence", 1))),
                    evidence=(
                        _pointer(
                            ledger="transfer_assessment",
                            event_id=assessment_id,
                            session_id=row.get("session_id"),
                        ),
                    ),
                    counterevidence=tuple(
                        str(value) for value in payload.get("blockers", [])
                    ),
                ),
            )
        )

    practice_groups: dict[tuple[UUID, str], list[Mapping[str, Any]]] = defaultdict(list)
    for row in practice_rows:
        key = (UUID(str(row["learner_id"])), str(row["competency_id"]))
        practice_groups[key].append(row)
    for (learner_id, competency_id), rows in practice_groups.items():
        ordered = sorted(
            rows, key=lambda item: int(_value(item, "sequence_no", 0)), reverse=True
        )
        recent = [item for item in ordered if item.get("outcome") == "needs_retry"][:2]
        if len(recent) < 2:
            continue
        newest = recent[0]
        evidence = tuple(
            _pointer(
                ledger="practice_attempt",
                event_id=item["attempt_record_id"],
                session_id=item.get("session_id"),
            )
            for item in recent
        )
        output.append(
            DerivedSignal(
                learner_id=learner_id,
                competency_id=competency_id,
                signal=LearnerAttentionSignal(
                    signal_id=_signal_id(
                        "practice", newest["attempt_record_id"], "stagnation"
                    ),
                    learner_ref=_learner_ref(learner_id),
                    signal_type="growth_stagnation",
                    severity="moderate",
                    state="monitoring",
                    uncertainty=max(
                        float(_value(item, "uncertainty", 1.0)) for item in recent
                    ),
                    observed_sequence=max(1, int(_value(newest, "sequence_no", 1))),
                    evidence=evidence,
                    counterevidence=tuple(
                        str(value)
                        for item in recent
                        for value in _value(item, "counterevidence", [])
                    ),
                ),
            )
        )

    return tuple(
        sorted(
            output,
            key=lambda item: (item.signal.learner_ref, item.signal.signal_id),
        )
    )


async def _load_attention_rows(
    conn: asyncpg.Connection, cohort_id: str
) -> dict[str, Sequence[Mapping[str, Any]]]:
    trajectory = await conn.fetch(
        """
        SELECT DISTINCT ON (r.learner_id, r.case_id)
          r.learner_id, r.revision_id, r.anchor_session_id, r.revision_no, r.assessment
        FROM app.outcome_trajectory_revision r
        JOIN app.app_user u ON u.user_id = r.learner_id
        WHERE u.cohort = $1
        ORDER BY r.learner_id, r.case_id, r.revision_no DESC
        """,
        cohort_id,
    )
    ruptures = await conn.fetch(
        """
        SELECT DISTINCT ON (ep.learner_id, ev.episode_id)
          ep.learner_id, ev.observation_id, ev.session_id, ev.sequence_no,
          ev.to_state, ev.uncertainty
        FROM app.rupture_observation_event ev
        JOIN app.rupture_episode ep ON ep.episode_id = ev.episode_id
        JOIN app.app_user u ON u.user_id = ep.learner_id
        WHERE u.cohort = $1
        ORDER BY ep.learner_id, ev.episode_id, ev.sequence_no DESC
        """,
        cohort_id,
    )
    safety = await conn.fetch(
        """
        SELECT se.id AS safety_event_id, s.learner_id, se.session_id,
               se.turn_id, se.trigger_type, se.ko_risk_level, se.escalated,
               row_number() OVER (
                 PARTITION BY s.learner_id ORDER BY se.created_at, se.id
               )::int AS observed_sequence
        FROM app.safety_events se
        JOIN app.sessions s ON s.id = se.session_id
        JOIN app.app_user u ON u.user_id = s.learner_id
        WHERE u.cohort = $1 AND se.escalated = TRUE
        ORDER BY s.learner_id, se.created_at, se.id
        """,
        cohort_id,
    )
    calibration = await conn.fetch(
        """
        SELECT * FROM (
          SELECT a.learner_id, a.competency_id, a.assessment_snapshot_id,
                 a.session_id, a.snapshot_no, a.assessment_payload,
                 row_number() OVER (
                   PARTITION BY a.learner_id, a.competency_id ORDER BY a.snapshot_no DESC
                 ) AS recent_rank
          FROM app.calibration_assessment_snapshot a
          JOIN app.app_user u ON u.user_id = a.learner_id
          WHERE u.cohort = $1
        ) ranked WHERE recent_rank <= 2
        """,
        cohort_id,
    )
    transfers = await conn.fetch(
        """
        SELECT DISTINCT ON (a.learner_id, a.competency_id)
          a.learner_id, a.competency_id, a.transfer_assessment_id,
          a.session_id, a.assessment_payload, 1 AS observed_sequence
        FROM app.calibration_transfer_assessment a
        JOIN app.app_user u ON u.user_id = a.learner_id
        WHERE u.cohort = $1
        ORDER BY a.learner_id, a.competency_id, a.created_at DESC
        """,
        cohort_id,
    )
    practices = await conn.fetch(
        """
        SELECT * FROM (
          SELECT a.learner_id, p.competency_id, a.attempt_record_id,
                 a.session_id, a.sequence_no, a.outcome, a.uncertainty,
                 a.counterevidence,
                 row_number() OVER (
                   PARTITION BY a.learner_id, p.competency_id ORDER BY a.created_at DESC
                 ) AS recent_rank
          FROM app.practice_attempt_evidence a
          JOIN app.practice_episode_submission e
            ON e.episode_submission_id = a.episode_submission_id
          JOIN app.practice_prescription p
            ON p.prescription_record_id = e.prescription_record_id
          JOIN app.app_user u ON u.user_id = a.learner_id
          WHERE u.cohort = $1 AND a.outcome = 'needs_retry'
        ) ranked WHERE recent_rank <= 2
        """,
        cohort_id,
    )
    return {
        "trajectory_rows": trajectory,
        "rupture_rows": ruptures,
        "safety_rows": safety,
        "calibration_rows": calibration,
        "transfer_rows": transfers,
        "practice_rows": practices,
    }


_MANIFEST_QUERIES: Mapping[str, tuple[str, str, str, str]] = {
    "alliance": (
        """
        SELECT m.measurement_id AS event_id, s.learner_id, m.session_id
        FROM app.measurement_event m
        JOIN app.sessions s ON s.id = m.session_id
        JOIN app.app_user u ON u.user_id = s.learner_id
        WHERE u.cohort = $1 AND m.construct = 'working_alliance' AND m.status = 'ready'
        ORDER BY m.created_at, m.measurement_id
        """,
        "measurement_event",
        "vignette.measurement-event.v1",
        "db://app.measurement_event",
    ),
    "rupture": (
        """
        SELECT ev.observation_id AS event_id, ep.learner_id, ev.session_id
        FROM app.rupture_observation_event ev
        JOIN app.rupture_episode ep ON ep.episode_id = ev.episode_id
        JOIN app.app_user u ON u.user_id = ep.learner_id
        WHERE u.cohort = $1
        ORDER BY ev.created_at, ev.observation_id
        """,
        "rupture_observation_event",
        "vignette.rupture-observation.v1",
        "db://app.rupture_observation_event",
    ),
    "transfer": (
        """
        SELECT a.transfer_assessment_id AS event_id, a.learner_id, a.session_id
        FROM app.calibration_transfer_assessment a
        JOIN app.app_user u ON u.user_id = a.learner_id
        WHERE u.cohort = $1
        ORDER BY a.created_at, a.transfer_assessment_id
        """,
        "transfer_assessment",
        "vignette.transfer-assessment.v1",
        "db://app.calibration_transfer_assessment",
    ),
    "calibration": (
        """
        SELECT a.assessment_snapshot_id AS event_id, a.learner_id, a.session_id
        FROM app.calibration_assessment_snapshot a
        JOIN app.app_user u ON u.user_id = a.learner_id
        WHERE u.cohort = $1
        ORDER BY a.created_at, a.assessment_snapshot_id
        """,
        "calibration_assessment",
        "vignette.calibration-assessment.v1",
        "db://app.calibration_assessment_snapshot",
    ),
}


def derive_manifest_input(
    rows_by_domain: Mapping[str, Sequence[Mapping[str, Any]]],
) -> ManifestInput | None:
    if set(rows_by_domain) != set(_MANIFEST_QUERIES):
        return None
    artifacts: list[Phase3EvidenceArtifact] = []
    sources: dict[str, tuple[UUID, LedgerEvidencePointer]] = {}
    fingerprint_payload: dict[str, list[str]] = {}
    for domain in ("alliance", "rupture", "transfer", "calibration"):
        rows = rows_by_domain[domain]
        if not rows:
            return None
        ordered = sorted(rows, key=lambda row: str(row["event_id"]))
        event_ids = [str(row["event_id"]) for row in ordered]
        fingerprint_payload[domain] = event_ids
        digest = _canonical_hash(event_ids)
        _, ledger, schema_version, provenance_uri = _MANIFEST_QUERIES[domain]
        artifacts.append(
            Phase3EvidenceArtifact(
                domain=domain,
                artifact_id=f"oas-g6-artifact-{domain}-{digest[:16]}",
                schema_version=schema_version,
                content_sha256=digest,
                record_count=len(event_ids),
                provenance_uri=provenance_uri,
            )
        )
        anchor = ordered[0]
        sources[domain] = (
            UUID(str(anchor["learner_id"])),
            _pointer(
                ledger=ledger,
                event_id=anchor["event_id"],
                session_id=anchor.get("session_id"),
            ),
        )
    return ManifestInput(
        artifacts=tuple(artifacts),
        source_by_domain=sources,
        source_fingerprint=_canonical_hash(fingerprint_payload),
    )


async def _load_manifest_rows(
    conn: asyncpg.Connection, cohort_id: str
) -> dict[str, Sequence[Mapping[str, Any]]]:
    output: dict[str, Sequence[Mapping[str, Any]]] = {}
    for domain, (query, _, _, _) in _MANIFEST_QUERIES.items():
        output[domain] = await conn.fetch(query, cohort_id)
    return output


async def produce_supervision_cycle(
    conn: asyncpg.Connection,
    *,
    cohort_id: str,
) -> dict[str, Any]:
    """한 코호트의 원장 상태를 idempotent G6 파생 산출물로 투영한다."""

    if not cohort_id.strip():
        raise ValueError("cohort_id must not be blank")
    rows = await _load_attention_rows(conn, cohort_id)
    derived = derive_attention_signals(**rows)
    attention_result: dict[str, Any] | None = None
    gap_results: list[dict[str, Any]] = []
    if derived:
        attention_fingerprint = _canonical_hash(
            [item.signal.model_dump(mode="json") for item in derived]
        )
        attention_submission_id = uuid5(
            _PRODUCER_NAMESPACE,
            f"attention-submission:{cohort_id}:{attention_fingerprint}",
        )
        attention_result = await supervision_research_store.append_attention_snapshot(
            conn,
            submission_id=attention_submission_id,
            snapshot_id=uuid5(
                _PRODUCER_NAMESPACE,
                f"attention-snapshot:{cohort_id}:{attention_fingerprint}",
            ),
            cohort_id=cohort_id,
            signals=[item.signal for item in derived],
            learner_ids_by_ref={
                item.signal.learner_ref: item.learner_id for item in derived
            },
        )

        gap_groups: dict[tuple[str, str], list[DerivedSignal]] = defaultdict(list)
        gap_kind_by_signal = {
            "deterioration": "coverage",
            "unresolved_rupture": "rupture_repair",
            "persistent_overconfidence": "calibration",
            "growth_stagnation": "growth_stagnation",
            "transfer_failure": "transfer",
        }
        for item in derived:
            gap_kind = gap_kind_by_signal.get(item.signal.signal_type)
            if gap_kind is None:
                # 안전 사건은 attention queue 대상이지 교육과정 성과 집계가 아니다.
                continue
            gap_groups[(item.competency_id, gap_kind)].append(item)
        for (competency_id, gap_kind), items in sorted(gap_groups.items()):
            evidence = [(item.learner_id, item.signal.evidence[0]) for item in items]
            gap_fingerprint = _canonical_hash(
                [
                    competency_id,
                    gap_kind,
                    [item.signal.signal_id for item in items],
                ]
            )
            result = await supervision_research_store.append_curriculum_gap(
                conn,
                submission_id=uuid5(
                    _PRODUCER_NAMESPACE,
                    f"gap-submission:{cohort_id}:{gap_fingerprint}",
                ),
                gap_snapshot_id=uuid5(
                    _PRODUCER_NAMESPACE,
                    f"gap-snapshot:{cohort_id}:{gap_fingerprint}",
                ),
                cohort_id=cohort_id,
                competency_id=competency_id,
                gap_kind=gap_kind,
                status="observed"
                if len({item.learner_id for item in items}) >= 2
                else "monitoring",
                uncertainty=max(item.signal.uncertainty for item in items),
                affected_learner_count=len({item.learner_id for item in items}),
                evidence=evidence,
            )
            gap_results.append(result)

    manifest_input = derive_manifest_input(await _load_manifest_rows(conn, cohort_id))
    manifest_result: dict[str, Any] | None = None
    if manifest_input is not None:
        manifest_result = await supervision_research_store.append_phase3_manifest(
            conn,
            submission_id=uuid5(
                _PRODUCER_NAMESPACE,
                f"manifest-submission:{cohort_id}:{manifest_input.source_fingerprint}",
            ),
            manifest_id=uuid5(
                _PRODUCER_NAMESPACE,
                f"manifest:{cohort_id}:{manifest_input.source_fingerprint}",
            ),
            cohort_id=cohort_id,
            artifacts=manifest_input.artifacts,
            source_by_domain=manifest_input.source_by_domain,
        )

    return {
        "cohort_id": cohort_id,
        "derived_signal_count": len(derived),
        "attention_snapshot": attention_result,
        "curriculum_gaps": gap_results,
        "phase3_manifest": manifest_result,
        "raw_transcript_included": False,
        "clinical_claim_allowed": False,
    }


async def produce_all_active_cohorts_once() -> dict[str, Any]:
    """현재 활성 학습자 코호트를 한 번 순회한다.

    각 코호트는 독립 트랜잭션이다. 한 코호트의 손상된 포인터가 다른 코호트의
    파생 산출물을 롤백하지 않는다.
    """

    async with db.acquire(ai_view="research", ai_context=True) as conn:
        rows = await conn.fetch(
            """
            SELECT DISTINCT cohort
            FROM app.app_user
            WHERE role = 'learner' AND is_active AND cohort IS NOT NULL
              AND length(btrim(cohort)) > 0
            ORDER BY cohort
            """
        )
    cohorts = [str(row["cohort"]) for row in rows]
    completed = 0
    failed = 0
    for cohort_id in cohorts:
        try:
            async with db.acquire(
                cohort=cohort_id,
                ai_view="supervisor",
                ai_context=True,
            ) as conn:
                await produce_supervision_cycle(conn, cohort_id=cohort_id)
            completed += 1
        except asyncio.CancelledError:
            raise
        except Exception:
            failed += 1
            logger.exception("G6 원장 파생 cycle 실패: cohort=%s", cohort_id)

    # Repo-approved synthetic evaluator comparison is a global research ledger,
    # not a learner cohort score.  Give it its own transaction so a malformed or
    # incomplete benchmark anchor can never roll back a completed supervisor cycle.
    version_comparison: dict[str, Any] = {
        "status": "skipped",
        "reason": "not_attempted",
        "raw_transcript_included": False,
        "clinical_claim_allowed": False,
    }
    version_comparison_failed = 0
    try:
        async with db.acquire(ai_view="research", ai_context=True) as conn:
            version_comparison = await supervision_research_version_evaluator.produce_repository_version_comparison(
                conn
            )
    except asyncio.CancelledError:
        raise
    except Exception:
        version_comparison_failed = 1
        version_comparison = {
            "status": "error",
            "reason": "repository_version_comparison_failed",
            "raw_transcript_included": False,
            "clinical_claim_allowed": False,
        }
        logger.exception("G6 repo benchmark version comparison 실패 격리")
    return {
        "cohort_count": len(cohorts),
        "completed": completed,
        "failed": failed,
        "version_comparison": version_comparison,
        "version_comparison_failed": version_comparison_failed,
    }


async def _producer_loop() -> None:
    delay = settings.supervision_research_producer_startup_delay_seconds
    if delay:
        await asyncio.sleep(delay)
    while True:
        try:
            result = await produce_all_active_cohorts_once()
            if result["cohort_count"]:
                logger.info(
                    "G6 원장 파생 cycle 완료: cohorts=%d completed=%d failed=%d "
                    "version_comparison=%s comparison_failed=%d",
                    result["cohort_count"],
                    result["completed"],
                    result["failed"],
                    result["version_comparison"].get("status"),
                    result["version_comparison_failed"],
                )
        except asyncio.CancelledError:
            raise
        except Exception:
            logger.exception("G6 원장 파생 scheduler cycle 실패")
        await asyncio.sleep(settings.supervision_research_producer_interval_seconds)


def schedule_supervision_research_producer() -> asyncio.Task[None] | None:
    global _PRODUCER_TASK
    if not settings.supervision_research_producer_enabled:
        return None
    if _PRODUCER_TASK is not None and not _PRODUCER_TASK.done():
        return _PRODUCER_TASK
    _PRODUCER_TASK = asyncio.create_task(
        _producer_loop(),
        name="supervision-research-ledger-producer",
    )
    return _PRODUCER_TASK


async def stop_supervision_research_producer() -> None:
    global _PRODUCER_TASK
    task = _PRODUCER_TASK
    _PRODUCER_TASK = None
    if task is None or task.done():
        return
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass


__all__ = [
    "DerivedSignal",
    "ManifestInput",
    "derive_attention_signals",
    "derive_manifest_input",
    "produce_all_active_cohorts_once",
    "produce_supervision_cycle",
    "schedule_supervision_research_producer",
    "stop_supervision_research_producer",
]