Phronesis

"""phronesis.core — typed primitives at the substrate layer.

This package contains:
- types: atomic primitives (Citation, Score, Finding, Recommendation) and
  deliverable aggregates (UseCase, MaturityDimension, etc.)
- engagement: state model + journal replay (M3)
- stages: stage transitions and review gates (M2)
- linter: P3/P7/P8/L1-L5 enforcement (M3)
- renderer: typed objects → markdown via Jinja2 templates (M3)
- selector: framework_selector primitive (M1)
- extraction: anonymizer + extraction pipeline (M7)

Imports are deferred to subpackages — keep this module empty.
"""

__version__ = "0.1.0-pre"

"""Tenant data anonymization for cross-engagement learning.

Per design spec §7.2 (AnonymizationPolicy), engagement extracts that flow
back into the knowledge graph (research/entities/) MUST be stripped of
tenant-identifying markers. The output should read like Bloomberg
"industry color" — preserves learning, identity-free.

This module ships the policy + the redaction function. The
test_anonymization_canary.py test (Phase E.3) is the release gate: any
checkin that lets a tenant marker leak into anonymized output blocks the
push.

Phase 1 implementation: regex-driven, conservative bias toward over-redaction
(false positives are recoverable; false negatives leak tenant data).
Phase 2+ may swap in NER-based name detection once a real engagement
provides ground truth.
"""

from __future__ import annotations

import re
from decimal import Decimal

from pydantic import BaseModel, Field

from core.types import TenantContext


class AnonymizationPolicy(BaseModel):
    """Configuration for the anonymization pass.

    Defaults to strict-by-default: every privacy-relevant transform is on.
    Callers explicitly opt out via False if they need raw text (e.g. for
    a tenant-private archive snapshot).
    """

    strip_tenant_slug: bool = True
    strip_tenant_name: bool = True
    strip_personal_names: bool = True
    replace_currency_with_bands: bool = True
    replace_dates_with_relative: bool = True
    replace_locations_with_regions: bool = True
    redact_terms: list[str] = Field(default_factory=list)


# ----------------------------------------------------------------------------
# Currency band mapping
# ----------------------------------------------------------------------------


_CURRENCY_BANDS: list[tuple[Decimal, str]] = [
    (Decimal("1000"), "sub-low-4-figures USD"),
    (Decimal("10000"), "low-4-figures USD"),
    (Decimal("100000"), "mid-5-figures USD"),
    (Decimal("1000000"), "low-6-figures USD"),
    (Decimal("10000000"), "low-7-figures USD"),
    (Decimal("100000000"), "low-8-figures USD"),
    (Decimal("1000000000"), "low-9-figures USD"),
]


def _currency_band(amount: Decimal) -> str:
    """Bucket a Decimal amount into a Bloomberg-style band string."""
    for ceiling, label in _CURRENCY_BANDS:
        if amount < ceiling:
            return label
    return "9+-figures USD"


_CURRENCY_RE = re.compile(
    r"\$\s?([0-9][0-9,]*(?:\.[0-9]+)?)(?:\s?([KMB]))?",
    re.IGNORECASE,
)


def _replace_currency(match: re.Match[str]) -> str:
    raw = match.group(1).replace(",", "")
    suffix = (match.group(2) or "").upper()
    multiplier = {"K": Decimal("1e3"), "M": Decimal("1e6"), "B": Decimal("1e9")}.get(
        suffix, Decimal("1")
    )
    try:
        amount = Decimal(raw) * multiplier
    except (ArithmeticError, ValueError):
        return match.group(0)
    return _currency_band(amount)


# ----------------------------------------------------------------------------
# Date relative-replacement
# ----------------------------------------------------------------------------


_ISO_DATE_RE = re.compile(r"\b(\d{4})-(\d{2})-(\d{2})\b")
_QUARTER_RE = re.compile(r"\b(\d{4})-Q[1-4]\b")
_YEAR_RE = re.compile(r"\b(20[2-3]\d)\b")  # 2020-2039


def _replace_iso_date(match: re.Match[str]) -> str:
    return "<engagement-date>"


def _replace_quarter(match: re.Match[str]) -> str:
    return "<engagement-quarter>"


def _replace_year(match: re.Match[str]) -> str:
    return "<engagement-year>"


# ----------------------------------------------------------------------------
# Personal-name detection (Phase 1: heuristic — capitalized adjacent words
# with at least one of: a Latin diacritic-bearing char, an honorific prefix,
# or a hyphen). False-negative prone — Phase 2 swaps in spaCy NER.
# ----------------------------------------------------------------------------


_NAME_RE = re.compile(
    r"\b("
    r"(?:Sr\.|Sra\.|Dr\.|Dra\.|Mr\.|Ms\.|Mrs\.)\s+[A-ZÁÉÍÓÚÑÜ][a-záéíóúñü]+"
    r"(?:\s+[A-ZÁÉÍÓÚÑÜ][a-záéíóúñü]+)*"
    r"|"
    r"[A-ZÁÉÍÓÚÑÜ][a-záéíóúñü]+(?:\s+[A-ZÁÉÍÓÚÑÜ][a-záéíóúñü]+)+"
    r")"
)


# Tokens that look like names but are domain language we want to keep.
# Phase 1 scope: keep frameworks + roles + Latin geographies that aren't
# tenant-specific.
_NAME_ALLOWLIST: set[str] = {
    # Frameworks
    "Wardley Mapping",
    "Three Horizons",
    "Real Options",
    "Value Proposition Canvas",
    "Jobs To Be Done",
    "MIT CISR",
    "Andrew Ng Pipeline",
    "QuantumBlack ML",
    "Lean Canvas",
    "Business Model Canvas",
    "Cost Of Delay",
    "Owner Earnings",
    "Microsoft Responsible AI",
    "Google PAIR",
    "Five Forces",
    "Where To Play How To Win",
    "Forrester Data",
    "Mckinsey 7S",
    "Mckinsey Influence",
    "Sean Ellis",
    "Sean McBride",
    # Roles (left as-is; the role itself isn't identifying)
    "Head Of Operations",
    "Head Of Customer Service",
    "Head Of Data",
    "Commercial Director",
    "Lead Data Engineer",
    # Regions / public infrastructure (intentionally generic)
    "LATAM",
    "United States",
    "South America",
}


def _replace_name(match: re.Match[str], policy: AnonymizationPolicy) -> str:
    raw = match.group(1)
    if raw in _NAME_ALLOWLIST:
        return raw
    return "<person>"


# ----------------------------------------------------------------------------
# Location detection (Phase 1: explicit list of CO/LATAM cities + country names).
# ----------------------------------------------------------------------------


_LOCATIONS: list[str] = [
    # Colombian cities likely to appear in early engagements
    "Bogotá",
    "Bogota",
    "Medellín",
    "Medellin",
    "Cali",
    "Barranquilla",
    "Cartagena",
    "Bucaramanga",
    # Country names commonly in scope
    "Colombia",
    "Mexico",
    "Brasil",
    "Brazil",
    "Argentina",
    "Chile",
    "Perú",
    "Peru",
]


def _strip_locations(text: str) -> str:
    """Replace specific city/country names with generic LATAM placeholder."""
    pattern = r"\b(" + "|".join(re.escape(loc) for loc in _LOCATIONS) + r")\b"
    return re.sub(pattern, "<latam-region>", text)


# ----------------------------------------------------------------------------
# Public entry point
# ----------------------------------------------------------------------------


def anonymize(
    text: str,
    policy: AnonymizationPolicy,
    tenant: TenantContext,
) -> str:
    """Apply the anonymization policy to `text` for the given tenant.

    Operations applied (in order):
      1. Strip tenant slug + name (literal substring replacement).
      2. Strip caller-supplied redact_terms.
      3. Replace personal names with <person> token.
      4. Replace location names with <latam-region>.
      5. Replace ISO dates / YYYY-Q* / years with relative tokens.
      6. Replace $-prefixed currency with magnitude bands.

    Order matters: dates run before currency because ISO dates contain
    digits that the currency regex would otherwise match-and-corrupt.
    """
    out = text

    if policy.strip_tenant_slug and tenant.tenant_slug:
        out = out.replace(tenant.tenant_slug, "<tenant>")
    if policy.strip_tenant_name and tenant.name:
        out = out.replace(tenant.name, "<tenant>")

    for term in policy.redact_terms:
        if term:
            out = out.replace(term, "<redacted>")

    if policy.strip_personal_names:
        # Strip tenant sponsor name explicitly (more reliable than regex)
        if tenant.sponsor:
            out = out.replace(tenant.sponsor, "<person>")
        out = _NAME_RE.sub(lambda m: _replace_name(m, policy), out)

    if policy.replace_locations_with_regions:
        out = _strip_locations(out)

    if policy.replace_dates_with_relative:
        out = _ISO_DATE_RE.sub(_replace_iso_date, out)
        out = _QUARTER_RE.sub(_replace_quarter, out)
        out = _YEAR_RE.sub(_replace_year, out)

    if policy.replace_currency_with_bands:
        out = _CURRENCY_RE.sub(_replace_currency, out)

    return out


def carries_tenant_marker(text: str, tenant: TenantContext) -> list[str]:
    """Forensic helper: return a list of tenant-identifying markers found
    in `text`. Empty list = clean. Used by the canary test to produce a
    descriptive failure message rather than a bare assertion.
    """
    findings: list[str] = []

    if tenant.tenant_slug and tenant.tenant_slug in text:
        findings.append(f"tenant_slug:{tenant.tenant_slug}")
    if tenant.name and tenant.name in text:
        findings.append(f"tenant_name:{tenant.name}")
    if tenant.sponsor and tenant.sponsor in text:
        findings.append(f"sponsor:{tenant.sponsor}")

    return findings


__all__ = [
    "AnonymizationPolicy",
    "anonymize",
    "carries_tenant_marker",
]

"""Engagement aggregate root + journal replay + JSONL persistence.

Closes Gap #12 (EngagementJournal replay) + Gap #13 (Engagement aggregate root)
from the Tropico Renovables synthetic engagement (2026-05-06).

EngagementJournal is the persistence boundary — append-only event log.
EngagementState is derived by replaying the journal — never mutated directly.
Engagement is the aggregate root: tenant + journal + emit().

JSONL persistence (Phase E):
- save_jsonl(path) writes one line per event (newline-delimited JSON).
- load_jsonl(tenant, path) reconstructs the journal. Missing path → empty.
- Phase 1 rewrites the whole file on save; Phase 3 will switch to true
  append-only writes once we move to lago.

Phase 3 mapping: this struct mirrors lago's `Aggregate<E, S>` shape, where
events becomes Vec<Record<E>> and replay() is the fold (`events.iter().fold(S::initial(), apply)`).
"""

from __future__ import annotations

from pathlib import Path
from typing import Literal

from pydantic import BaseModel, Field

from core.types import EventKind, JournalEvent, TenantContext


class EngagementState(BaseModel):
    """Derived state from replaying an EngagementJournal. Read-only.

    Stages mutate engagement state via emit() — never by setting fields here.
    EngagementState is rebuilt fresh on every replay() call.
    """

    current_stage: Literal["intake", "scan", "ideate", "prioritize", "roadmap", "concluded"] = (
        "intake"
    )
    thesis_id: str | None = None
    frameworks_active: list[str] = Field(default_factory=list)
    maturity_dimensions: list[str] = Field(default_factory=list)
    use_cases: dict[str, dict[str, object]] = Field(default_factory=dict)
    use_cases_prioritized: list[str] = Field(default_factory=list)
    baselines_captured: list[str] = Field(default_factory=list)
    deliverables_rendered: list[str] = Field(default_factory=list)
    review_pending: str | None = None
    is_concluded: bool = False


class EngagementJournal(BaseModel):
    """Append-only event log scoped to one tenant engagement.

    Persistence path: engagements/<tenant_slug>/journal.jsonl (P6 — gitignored).
    """

    tenant: TenantContext
    events: list[JournalEvent] = Field(default_factory=list)

    def append(self, event: JournalEvent) -> None:
        """Append an event. Caller is responsible for ULID monotonicity
        (which Engagement.emit() handles via JournalEvent's default_factory)."""
        self.events.append(event)

    def replay(self) -> EngagementState:
        """Reconstruct EngagementState by folding events in journal order.

        Pure function — calling .replay() twice returns equal states.
        Phase 3 mirror: events.iter().fold(EngagementState::default(), apply).
        """
        state = EngagementState()
        for ev in self.events:
            state = _apply(state, ev)
        return state

    def save_jsonl(self, path: Path) -> None:
        """Write the journal to `path` as newline-delimited JSON.

        One event per line. Caller chooses where to write (typically
        `engagements/<tenant_slug>/journal.jsonl`). Parent dirs created
        if missing. Existing file is overwritten — the journal is the
        source of truth, and every emit() rewrites the whole file.

        Phase 3 will switch to true append-only writes; Phase 1 simplicity
        preferred for synthetic-fixture-driven workflows.
        """
        path.parent.mkdir(parents=True, exist_ok=True)
        lines = [ev.model_dump_json(exclude_none=True) for ev in self.events]
        path.write_text("\n".join(lines) + ("\n" if lines else ""))

    @classmethod
    def load_jsonl(cls, tenant: TenantContext, path: Path) -> EngagementJournal:
        """Reconstruct an EngagementJournal from an on-disk JSONL file.

        Missing file returns an empty journal — that's the natural startup
        case for a fresh engagement. Empty lines are skipped. Malformed
        lines raise ValidationError (don't silently drop events).
        """
        events: list[JournalEvent] = []
        if path.exists():
            for line in path.read_text().splitlines():
                stripped = line.strip()
                if not stripped:
                    continue
                events.append(JournalEvent.model_validate_json(stripped))
        return cls(tenant=tenant, events=events)


_STAGE_ORDER = ["intake", "scan", "ideate", "prioritize", "roadmap"]


def _next_stage(approved_stage: str) -> str:
    try:
        i = _STAGE_ORDER.index(approved_stage)
    except ValueError:
        return approved_stage
    if i + 1 < len(_STAGE_ORDER):
        return _STAGE_ORDER[i + 1]
    return "concluded"


def _apply(state: EngagementState, ev: JournalEvent) -> EngagementState:
    """Pure event-application. Returns a new state — never mutates input."""
    next_state = state.model_copy(deep=True)

    if ev.kind == EventKind.STRATEGIC_THESIS_DECLARED:
        thesis_id = ev.payload.get("thesis_id")
        if isinstance(thesis_id, str):
            next_state.thesis_id = thesis_id

    elif ev.kind == EventKind.INTAKE_COMPLETED:
        fw = ev.payload.get("frameworks_selected", [])
        if isinstance(fw, list):
            next_state.frameworks_active = [str(x) for x in fw]
        # current_stage transitions on STAGE_REVIEW_APPROVED, not here.

    elif ev.kind == EventKind.MATURITY_DIMENSION_SCORED:
        name = ev.payload.get("dimension_name")
        if isinstance(name, str):
            next_state.maturity_dimensions.append(name)

    elif ev.kind == EventKind.USE_CASE_PROPOSED:
        uc_id = ev.payload.get("use_case_id")
        if isinstance(uc_id, str):
            next_state.use_cases[uc_id] = dict(ev.payload)

    elif ev.kind == EventKind.USE_CASE_PRIORITIZED:
        uc_id = ev.payload.get("use_case_id")
        if isinstance(uc_id, str):
            next_state.use_cases_prioritized.append(uc_id)

    elif ev.kind == EventKind.BASELINE_CAPTURED:
        m = ev.payload.get("metric_name")
        if isinstance(m, str):
            next_state.baselines_captured.append(m)

    elif ev.kind == EventKind.DELIVERABLE_RENDERED:
        slug = ev.payload.get("slug")
        if isinstance(slug, str):
            next_state.deliverables_rendered.append(slug)

    elif ev.kind == EventKind.STAGE_REVIEW_REQUESTED:
        st = ev.payload.get("stage")
        if isinstance(st, str):
            next_state.review_pending = st

    elif ev.kind == EventKind.STAGE_REVIEW_APPROVED:
        next_state.review_pending = None
        st = ev.payload.get("stage")
        if isinstance(st, str):
            next_state.current_stage = _next_stage(st)  # type: ignore[assignment]

    elif ev.kind == EventKind.ENGAGEMENT_CONCLUDED:
        next_state.is_concluded = True
        next_state.current_stage = "concluded"

    return next_state


class Engagement(BaseModel):
    """Aggregate root: tenant + journal + emit().

    Closes Gap #13 — the Tropico engagement had to manually couple
    TenantContext + journal + deliverables, losing invariants. Engagement
    forces all mutation through emit() so replay stays deterministic.
    """

    tenant: TenantContext
    journal: EngagementJournal

    def state(self) -> EngagementState:
        """Replay the journal to derive current state. Pure function."""
        return self.journal.replay()

    def emit(
        self,
        kind: EventKind,
        stage: str,
        payload: dict[str, object],
        actor: str = "phronesis",
        parent_event_id: str | None = None,
    ) -> str:
        """Emit a journal event. Returns the event_id (ULID).

        Construction goes through JournalEvent's model_validator (Phase A.3),
        which re-validates the payload against the kind's schema. Bad
        payloads raise ValidationError before append.

        Reflexive M7 hook: when `kind == ENGAGEMENT_CONCLUDED`, the
        extraction pipeline fires automatically (per
        `feedback_bookkeeping_reflexive.md`). Disable via env var
        `PHRONESIS_EXTRACTION_ENABLED=0` (interactive-debug use only —
        the default ON path is the autonomous-arc contract).
        """
        ev = JournalEvent(
            kind=kind,
            actor=actor,
            stage=stage,
            payload=payload,
            parent_event_id=parent_event_id,
        )
        self.journal.append(ev)

        if kind == EventKind.ENGAGEMENT_CONCLUDED:
            self._fire_extraction_hook()

        return ev.event_id

    def _fire_extraction_hook(self) -> None:
        """Trigger M7 extraction pipeline on ENGAGEMENT_CONCLUDED.

        Imported lazily to avoid a circular import (extraction depends on
        Engagement). Best-effort: any exception in the hook is swallowed
        with a stderr warning — the engagement journal is the source of
        truth, and the extraction pipeline must not block emit().
        """
        try:
            from core.extraction.pipeline import on_engagement_concluded

            on_engagement_concluded(self)
        except Exception as exc:  # pragma: no cover — hook is fire-and-forget
            import sys

            print(
                f"[phronesis][warn] extraction-hook failed for "
                f"{self.tenant.tenant_slug!r}: {exc!r}",
                file=sys.stderr,
            )

"""Typed payloads for each EventKind. Closes Gap #11 from the Tropico
Renovables synthetic engagement (2026-05-06).

JournalEvent.payload was originally `dict[str, object]` — accepted any dict.
The replay function (Phase A.4) and renderers (M3) need stable, typed field
shapes, so each EventKind now has a Pydantic class registered here. The
JournalEvent model_validator (Phase A.3) re-validates `payload` against the
declared kind's schema at construction.

Phase 3 mapping: this registry mirrors lago's `enum E { Started{...}, ... }`
shape for clean transcription to Rust.
"""

from __future__ import annotations

from typing import Literal

from pydantic import BaseModel

from core.types import EventKind


class EngagementStartedPayload(BaseModel):
    """ENGAGEMENT_STARTED — emitted exactly once when an Engagement is created."""

    tenant_slug: str
    scope: str
    sponsor: str
    target_duration_weeks: int


class IntakeCompletedPayload(BaseModel):
    """INTAKE_COMPLETED — Stage 1 closes after thesis declared + frameworks selected."""

    thesis_id: str
    frameworks_selected: list[str]


class InterviewLoggedPayload(BaseModel):
    """INTERVIEW_LOGGED — sponsor / SME interview captured during intake."""

    interviewee: str
    role: str
    transcript_ref: str
    key_findings: list[str]


class DocumentIngestedPayload(BaseModel):
    """DOCUMENT_INGESTED — internal doc consumed during intake (data, regs, prior reports)."""

    path: str
    kind: Literal["interview", "data", "regulatory", "report"]
    summary: str


class StrategicThesisDeclaredPayload(BaseModel):
    """STRATEGIC_THESIS_DECLARED — L1 milestone event."""

    thesis_id: str
    economic_lever: str
    lever_kind: str
    magnitude_estimate: str  # str-encoded Decimal for journal portability
    horizon: str
    owner: str


class MaturityDimensionScoredPayload(BaseModel):
    """MATURITY_DIMENSION_SCORED — emitted once per scored dimension in Stage 2."""

    dimension_name: str
    current_value: float
    target_value: float
    framework_ref: str
    gap_summary: str


class UseCaseProposedPayload(BaseModel):
    """USE_CASE_PROPOSED — surfaced during Stage 3 ideation."""

    use_case_id: str
    expected_value: str  # str-encoded Decimal
    cost_estimate: str
    ideation_source: str
    data_readiness_band: str


class UseCasePrioritizedPayload(BaseModel):
    """USE_CASE_PRIORITIZED — ranked + ROI-modeled in Stage 4."""

    use_case_id: str
    rice_score: float
    year1_net: str  # str-encoded Decimal
    rank: int


class RoadmapStepProposedPayload(BaseModel):
    """ROADMAP_STEP_PROPOSED — one step in Stage 5 Three-Horizons plan."""

    step_id: str
    horizon: str
    quarter: str
    owner: str
    success_gate: str


class BaselineCapturedPayload(BaseModel):
    """BASELINE_CAPTURED — L5 enforcement; must precede PILOT_STARTED."""

    metric_name: str
    baseline_value: str  # str-encoded Decimal
    captured_by: str
    measurement_date: str  # ISO 8601


class PilotStartedPayload(BaseModel):
    """PILOT_STARTED — L5 gate fires here; baseline must already be captured."""

    use_case_id: str
    pilot_design_ref: str
    start_date: str
    duration_weeks: int


class DeliverableRenderedPayload(BaseModel):
    """DELIVERABLE_RENDERED — emitted by render orchestrator (M3)."""

    slug: str
    output_path: str
    linter_passed: bool
    lint_warnings: list[str]


class StageReviewRequestedPayload(BaseModel):
    """STAGE_REVIEW_REQUESTED — sponsor gate. P5 enforcement."""

    stage: str
    reviewer: str
    summary: str
    artifacts: list[str]
    deadline: str | None = None


class StageReviewApprovedPayload(BaseModel):
    """STAGE_REVIEW_APPROVED — gate cleared, advances current_stage."""

    stage: str
    reviewer: str
    notes: str | None = None


class StageReviewRevisedPayload(BaseModel):
    """STAGE_REVIEW_REVISED — sponsor sends stage back. Triggers retract_to_revision_point (A.8)."""

    stage: str
    reviewer: str
    revisions_requested: list[str]
    original_event_id: str


class EngagementConcludedPayload(BaseModel):
    """ENGAGEMENT_CONCLUDED — terminal event. Stages all approved + deliverables rendered."""

    stages_approved: int
    thesis_id: str
    top_pilot: str
    deliverable_slugs: list[str]


_REGISTRY: dict[EventKind, type[BaseModel]] = {
    EventKind.ENGAGEMENT_STARTED: EngagementStartedPayload,
    EventKind.INTAKE_COMPLETED: IntakeCompletedPayload,
    EventKind.INTERVIEW_LOGGED: InterviewLoggedPayload,
    EventKind.DOCUMENT_INGESTED: DocumentIngestedPayload,
    EventKind.STRATEGIC_THESIS_DECLARED: StrategicThesisDeclaredPayload,
    EventKind.MATURITY_DIMENSION_SCORED: MaturityDimensionScoredPayload,
    EventKind.USE_CASE_PROPOSED: UseCaseProposedPayload,
    EventKind.USE_CASE_PRIORITIZED: UseCasePrioritizedPayload,
    EventKind.ROADMAP_STEP_PROPOSED: RoadmapStepProposedPayload,
    EventKind.BASELINE_CAPTURED: BaselineCapturedPayload,
    EventKind.PILOT_STARTED: PilotStartedPayload,
    EventKind.DELIVERABLE_RENDERED: DeliverableRenderedPayload,
    EventKind.STAGE_REVIEW_REQUESTED: StageReviewRequestedPayload,
    EventKind.STAGE_REVIEW_APPROVED: StageReviewApprovedPayload,
    EventKind.STAGE_REVIEW_REVISED: StageReviewRevisedPayload,
    EventKind.ENGAGEMENT_CONCLUDED: EngagementConcludedPayload,
}


def payload_for(kind: EventKind) -> type[BaseModel]:
    """Return the typed payload class for an EventKind.

    Raises KeyError if the kind has no registered payload schema. After
    Phase A.2, all 16 EventKinds are registered and this never raises in
    practice.
    """
    return _REGISTRY[kind]

"""Phronesis M7 — extraction pipeline.

Transforms a concluded engagement journal into anonymized knowledge-graph
candidates that flow through the bookkeeping P8 review queue and land as
entity pages in `research/entities/{industry-pattern,framework-refinement}/`.

Hook point: `ENGAGEMENT_CONCLUDED` event (per `feedback_bookkeeping_reflexive.md`)
triggers `extract_and_queue(engagement)`. Each emitted candidate carries:
  - anonymized text (per `core/anonymize.py::AnonymizationPolicy`),
  - candidate entity slug + type (industry-pattern / framework-refinement),
  - provenance pointing back to the engagement journal,
  - bookkeeping P8 score (novelty + specificity + relevance).

Candidates score ≥5/9 land in `research/entities/`. Below-threshold candidates
go to the review queue at `~/.config/phronesis/extraction-queue/` for human
inspection. We never file entities directly — every candidate flows through
the bookkeeping gate per the anti-pattern in `feedback_bookkeeping_reflexive.md`.

The 14-canary-token release gate runs against the anonymizer output via
`tests/integration/test_anonymization_canary.py`. A leaking token blocks
the push.
"""

from __future__ import annotations

from core.extraction.anonymizer import (
    EngagementAnonymizer,
    anonymize_engagement_text,
)
from core.extraction.candidates import (
    ExtractionCandidate,
    extract_framework_refinements,
    extract_industry_patterns,
)
from core.extraction.pipeline import (
    ENTITY_GRAPH_ROOT,
    EXTRACTION_QUEUE_ROOT,
    ExtractionResult,
    extract_and_queue,
    on_engagement_concluded,
)

__all__ = [
    "EXTRACTION_QUEUE_ROOT",
    "ENTITY_GRAPH_ROOT",
    "EngagementAnonymizer",
    "ExtractionCandidate",
    "ExtractionResult",
    "anonymize_engagement_text",
    "extract_and_queue",
    "extract_framework_refinements",
    "extract_industry_patterns",
    "on_engagement_concluded",
]

"""Engagement-shaped anonymization wrapper.

Thin adapter over `core/anonymize.py` that takes a `TenantContext` +
engagement-derived `redact_terms` (e.g. project codenames, interviewee
non-sponsor names) and produces canary-clean text suitable for the
knowledge-graph layer.

Why wrap `anonymize()`? The extraction pipeline always operates on the
same shape (engagement journal payloads + tenant context), and we want a
single place where the policy defaults + the engagement-specific
redact_terms come together. Callers in `pipeline.py` and the test
harness use this wrapper, not the lower-level `anonymize()` directly.

The 14-canary-token release gate
(`tests/integration/test_anonymization_canary.py`) drives the underlying
`core/anonymize.py::anonymize()` directly because the canary asserts the
policy itself works — not just our wrapper. The wrapper is a convenience
layer for the extraction pipeline.
"""

from __future__ import annotations

from collections.abc import Iterable

from pydantic import BaseModel

from core.anonymize import AnonymizationPolicy, anonymize, carries_tenant_marker
from core.engagement import Engagement
from core.types import TenantContext


class EngagementAnonymizer(BaseModel):
    """Tenant-bound anonymizer for engagement-derived text.

    Built once per engagement; reused across every candidate emitted by
    the extraction pipeline.

    Attributes:
        tenant: the engagement tenant — name + slug + sponsor stripped.
        policy: the anonymization policy (defaults to strict — every
            transform on, no caller-supplied redact_terms unless the
            engagement surfaces project codenames).
    """

    tenant: TenantContext
    policy: AnonymizationPolicy

    def redact(self, text: str) -> str:
        """Anonymize `text` against this engagement's tenant + policy."""
        return anonymize(text, self.policy, self.tenant)

    def carries_marker(self, text: str) -> list[str]:
        """Forensic — list tenant markers still present in `text`. Empty
        list means clean. Used by the pipeline to assert post-redaction
        before emitting a candidate."""
        return carries_tenant_marker(text, self.tenant)


def _collect_engagement_redact_terms(engagement: Engagement) -> list[str]:
    """Walk the engagement journal for project codenames + non-sponsor
    interviewee names that the strict policy might miss.

    These get added to `policy.redact_terms`. Conservative bias toward
    over-redaction — the canary catches leaks, not over-redaction.

    Heuristics (Phase 1):
      - `INTERVIEW_LOGGED.interviewee` → strip the human name (modulo the
        sponsor, which `core/anonymize.py` already handles).
      - `ROADMAP_STEP_PROPOSED.title` may contain project codenames —
        Phase 2 will add NER + per-tenant project-codename declarations.
        For now, we trust the strict policy + the canary list.
    """
    from core.types import EventKind

    redact: list[str] = []
    for ev in engagement.journal.events:
        if ev.kind == EventKind.INTERVIEW_LOGGED:
            interviewee = ev.payload.get("interviewee", "")
            if not isinstance(interviewee, str) or not interviewee:
                continue
            # Strip any "(Role)" suffix — we only want the name itself.
            name = interviewee.split("(")[0].strip()
            # Don't redact the sponsor here; `anonymize()` already does it.
            if engagement.tenant.sponsor and engagement.tenant.sponsor in name:
                continue
            if name and name not in redact:
                redact.append(name)

    return redact


def anonymize_engagement_text(
    text: str,
    engagement: Engagement,
    *,
    extra_redact_terms: Iterable[str] = (),
    policy: AnonymizationPolicy | None = None,
) -> str:
    """Anonymize `text` using engagement-derived redaction terms.

    The convenience entry point for one-shot redactions (e.g. extracting
    a single finding into an industry-pattern candidate). The pipeline
    builds an `EngagementAnonymizer` once and calls `.redact()` repeatedly
    — that's the hot path. This function is the cold path for ad-hoc
    callers.

    Args:
        text: the raw text to anonymize.
        engagement: the engagement the text was derived from (for tenant
            context + journal-derived redact terms).
        extra_redact_terms: caller-supplied terms in addition to the
            journal-derived ones (e.g. a project codename surfaced in a
            specific journal payload).
        policy: optional policy override; default is strict.

    Returns:
        Anonymized text with all tenant markers + journal-derived
        identifying names + caller-supplied terms redacted.
    """
    if policy is None:
        policy = AnonymizationPolicy()

    journal_terms = _collect_engagement_redact_terms(engagement)
    combined_terms = list(
        dict.fromkeys([*policy.redact_terms, *journal_terms, *extra_redact_terms])
    )

    effective_policy = policy.model_copy(update={"redact_terms": combined_terms})
    return anonymize(text, effective_policy, engagement.tenant)


__all__ = [
    "EngagementAnonymizer",
    "anonymize_engagement_text",
]

"""Candidate extraction from concluded engagement journals.

Two extractors:
  - `extract_industry_patterns()` — recurring patterns within an industry
    (e.g. "LATAM mid-market banks with no production ML score 2.0±0.3 on
    customer-facing AI maturity"). Source: maturity-dimension scores +
    findings + thesis economic_lever, grouped by `tenant.industry`.
  - `extract_framework_refinements()` — empirical adjustments to canonical
    frameworks discovered during the engagement (e.g. "RICE under-weights
    regulatory-pressure use cases in financial services — recommend +25%
    impact bonus for `regulatory-pressure` source"). Source: deltas between
    rendered ROI vs RICE rank, ideation_source distribution, observed lift
    on capability-heatmap target_score vs baseline assumptions.

Each candidate carries enough provenance to round-trip back to the
journal event(s) that produced it. We never include raw text without
anonymization — the anonymizer wrapper is applied before the candidate
ever leaves this module.

Phase 1 scope: rule-based extractors. Phase 2 will swap in LLM extractors
once a corpus of ≥3 concluded engagements per industry exists.
"""

from __future__ import annotations

from typing import Literal

from pydantic import BaseModel, Field

from core.engagement import Engagement
from core.extraction.anonymizer import EngagementAnonymizer
from core.types import EventKind

EntityType = Literal["industry-pattern", "framework-refinement"]


class ExtractionCandidate(BaseModel):
    """A pre-bookkeeping candidate for a knowledge-graph entity.

    Fields mirror the bookkeeping `RawItem` shape so adapter glue stays
    minimal — `pipeline.py::_to_raw_item()` builds a bookkeeping RawItem
    directly from the candidate.

    Attributes:
        slug: kebab-case proposed entity slug (e.g.
            `latam-mid-market-banking-customer-ai-gap`).
        entity_type: which `research/entities/<type>/` directory the entity
            lands in if the bookkeeping P8 score lets it through.
        content: anonymized body text (Bloomberg industry color — no
            tenant identity).
        quote: the source phrase that surfaced the pattern (anonymized).
        title: short human-readable title for the entity page.
        provenance_event_ids: ULIDs of the journal events that produced
            this candidate. Round-trippable.
        industry: the industry signature for industry-pattern candidates
            (`tenant.industry`). None for framework-refinement candidates.
        framework_ref: the canonical framework being refined
            (e.g. `framework:rice`). None for industry-pattern candidates.
        signals: structured payload — numeric deltas, observed-vs-expected
            ratios, etc. that the bookkeeping scorer can pick up.
    """

    slug: str
    entity_type: EntityType
    content: str
    quote: str
    title: str
    provenance_event_ids: list[str]
    industry: str | None = None
    framework_ref: str | None = None
    signals: dict[str, float | str] = Field(default_factory=dict)


# ----------------------------------------------------------------------------
# Industry-pattern extraction
# ----------------------------------------------------------------------------


def extract_industry_patterns(
    engagement: Engagement,
    anonymizer: EngagementAnonymizer,
) -> list[ExtractionCandidate]:
    """Surface industry-pattern candidates from a concluded engagement.

    Phase 1 rules:
      1. Each scored maturity dimension below 2.5 on the canonical 1-5
         scale becomes a candidate ("industry X dimension Y under-developed").
      2. The strategic-thesis economic_lever becomes one industry-aware
         candidate ("industry X tenants commonly recover value via Z").

    Returns at most `n_dimensions + 1` candidates. The bookkeeping scorer
    decides which survive into `research/entities/industry-pattern/`.
    """
    candidates: list[ExtractionCandidate] = []
    industry = engagement.tenant.industry

    # Rule 1 — dimension under-development across industry.
    for ev in engagement.journal.events:
        if ev.kind != EventKind.MATURITY_DIMENSION_SCORED:
            continue
        dim_name_raw = ev.payload.get("dimension_name", "")
        current_raw = ev.payload.get("current_value", 0.0)
        target_raw = ev.payload.get("target_value", 0.0)
        gap_summary_raw = ev.payload.get("gap_summary", "")
        if not isinstance(dim_name_raw, str) or not dim_name_raw:
            continue
        try:
            current = float(current_raw) if isinstance(current_raw, (int, float, str)) else 0.0
            target = float(target_raw) if isinstance(target_raw, (int, float, str)) else 0.0
        except (TypeError, ValueError):
            continue

        if current >= 2.5:
            # Above-threshold dimensions don't surface as under-development
            # patterns; they're industry-baseline, not industry-pattern.
            continue

        gap_summary = gap_summary_raw if isinstance(gap_summary_raw, str) else ""

        body = (
            f"Industry pattern observed in {industry}: "
            f"dimension '{dim_name_raw}' scores {current:.1f}/5 against "
            f"benchmark target {target:.1f}. Observed gap: {gap_summary}. "
            f"Pattern recurs across same-industry tenants — surfaced once, "
            f"promotable when ≥3 industry-matched engagements reproduce."
        )
        quote = f"{dim_name_raw} at {current:.1f}/5: {gap_summary}"

        candidates.append(
            ExtractionCandidate(
                slug=_industry_pattern_slug(industry, dim_name_raw),
                entity_type="industry-pattern",
                content=anonymizer.redact(body),
                quote=anonymizer.redact(quote),
                title=f"{industry}: {dim_name_raw} under-development",
                provenance_event_ids=[ev.event_id],
                industry=industry,
                signals={
                    "dimension": dim_name_raw,
                    "current_score": current,
                    "target_score": target,
                    "gap_magnitude": target - current,
                },
            )
        )

    # Rule 2 — industry-aware economic lever pattern.
    thesis_ev = _find_event(engagement, EventKind.STRATEGIC_THESIS_DECLARED)
    if thesis_ev is not None:
        lever_raw = thesis_ev.payload.get("economic_lever", "")
        lever_kind_raw = thesis_ev.payload.get("lever_kind", "")
        magnitude_raw = thesis_ev.payload.get("magnitude_estimate", "")
        lever = lever_raw if isinstance(lever_raw, str) else ""
        lever_kind = lever_kind_raw if isinstance(lever_kind_raw, str) else ""
        magnitude = magnitude_raw if isinstance(magnitude_raw, str) else "0"

        if lever:
            body = (
                f"Industry pattern observed in {industry}: tenants surface "
                f"a {lever_kind} economic lever — '{lever}'. Magnitude "
                f"estimate (anonymized to band): see signals. Surfaced once; "
                f"promotion to industry-pattern entity requires ≥3 same-"
                f"industry engagements with the same lever_kind."
            )
            candidates.append(
                ExtractionCandidate(
                    slug=_industry_lever_slug(industry, lever_kind),
                    entity_type="industry-pattern",
                    content=anonymizer.redact(body),
                    quote=anonymizer.redact(lever),
                    title=f"{industry}: {lever_kind}-lever pattern",
                    provenance_event_ids=[thesis_ev.event_id],
                    industry=industry,
                    signals={
                        "lever_kind": lever_kind,
                        "magnitude_band_raw": magnitude,
                    },
                )
            )

    return candidates


# ----------------------------------------------------------------------------
# Framework-refinement extraction
# ----------------------------------------------------------------------------


def extract_framework_refinements(
    engagement: Engagement,
    anonymizer: EngagementAnonymizer,
) -> list[ExtractionCandidate]:
    """Surface framework-refinement candidates from a concluded engagement.

    Phase 1 rules:
      1. If 2+ use cases share the same `ideation_source`, surface a RICE
         refinement candidate ("source X over-represented in industry Y —
         consider weighting").
      2. If the top-ranked use case by RICE differs from the top by year1_net
         ROI, surface a delta candidate ("RICE rank vs ROI rank diverges
         for industry Y — calibrate impact weighting").
    """
    candidates: list[ExtractionCandidate] = []
    industry = engagement.tenant.industry

    # Gather use-case events with ideation source + RICE rank + ROI.
    proposed: dict[str, dict[str, object]] = {}
    prioritized: list[dict[str, object]] = []

    for ev in engagement.journal.events:
        if ev.kind == EventKind.USE_CASE_PROPOSED:
            uc_id = ev.payload.get("use_case_id")
            if isinstance(uc_id, str):
                proposed[uc_id] = {**ev.payload, "_event_id": ev.event_id}
        elif ev.kind == EventKind.USE_CASE_PRIORITIZED:
            uc_id = ev.payload.get("use_case_id")
            if isinstance(uc_id, str):
                prioritized.append({**ev.payload, "_event_id": ev.event_id})

    # Rule 1 — ideation source over-representation in same engagement.
    sources: dict[str, list[str]] = {}
    source_event_ids: dict[str, list[str]] = {}
    for uc_id, pl in proposed.items():
        source_raw = pl.get("ideation_source", "")
        if not isinstance(source_raw, str) or not source_raw:
            continue
        sources.setdefault(source_raw, []).append(uc_id)
        ev_id = pl.get("_event_id")
        if isinstance(ev_id, str):
            source_event_ids.setdefault(source_raw, []).append(ev_id)

    for source, uc_ids in sources.items():
        if len(uc_ids) < 2:
            continue
        body = (
            f"Framework refinement candidate: in this {industry} engagement, "
            f"{len(uc_ids)} of {len(proposed)} use cases originated from "
            f"ideation source '{source}'. RICE assumes uniform-priority "
            f"sourcing; over-represented sources may need explicit "
            f"weighting in same-industry calibrations. Surfaced once; "
            f"promotion to framework-refinement entity requires ≥3 "
            f"engagements showing the same source skew."
        )
        candidates.append(
            ExtractionCandidate(
                slug=_framework_rice_source_slug(industry, source),
                entity_type="framework-refinement",
                content=anonymizer.redact(body),
                quote=anonymizer.redact(f"{len(uc_ids)}/{len(proposed)} use cases from '{source}'"),
                title=f"RICE: {source} weighting in {industry}",
                provenance_event_ids=source_event_ids.get(source, []),
                framework_ref="framework:rice",
                signals={
                    "ideation_source": source,
                    "use_case_count": float(len(uc_ids)),
                    "total_proposed": float(len(proposed)),
                    "share": float(len(uc_ids)) / float(len(proposed) or 1),
                },
            )
        )

    # Rule 2 — RICE rank vs ROI year1_net divergence at rank 1.
    if prioritized:
        try:
            top_rice_uc = min(
                prioritized,
                key=_rank_key,
            )
            # Top by year1_net = max
            top_roi_uc = max(
                prioritized,
                key=lambda p: _decimal_str_to_float(p.get("year1_net", "0")),
            )
        except (TypeError, ValueError):
            top_rice_uc = None
            top_roi_uc = None

        if (
            top_rice_uc is not None
            and top_roi_uc is not None
            and top_rice_uc.get("use_case_id") != top_roi_uc.get("use_case_id")
        ):
            ev_ids = [
                eid
                for eid in (
                    _event_id_or_none(top_rice_uc),
                    _event_id_or_none(top_roi_uc),
                )
                if eid is not None
            ]
            body = (
                f"Framework refinement candidate: in this {industry} "
                f"engagement, RICE top-ranked use case differs from ROI "
                f"year-1-net top. RICE's reach×impact×confidence/effort "
                f"calculation may under-weight near-term revenue capture "
                f"in this industry. Surfaced once; promotion requires ≥3 "
                f"same-industry engagements with the same RICE/ROI rank "
                f"divergence."
            )
            candidates.append(
                ExtractionCandidate(
                    slug=_framework_rice_roi_slug(industry),
                    entity_type="framework-refinement",
                    content=anonymizer.redact(body),
                    quote=anonymizer.redact(f"RICE-top vs ROI-top diverge in {industry}"),
                    title=f"RICE: RICE-vs-ROI rank gap in {industry}",
                    provenance_event_ids=list(dict.fromkeys(ev_ids)),
                    framework_ref="framework:rice",
                    signals={
                        "industry": industry,
                        "rice_top_use_case": str(top_rice_uc.get("use_case_id", "")),
                        "roi_top_use_case": str(top_roi_uc.get("use_case_id", "")),
                    },
                )
            )

    return candidates


# ----------------------------------------------------------------------------
# Slug builders
# ----------------------------------------------------------------------------


def _industry_pattern_slug(industry: str, dimension: str) -> str:
    """Build a deterministic slug for an industry-dimension pattern.

    Format: `<industry>-<dimension-kebab>-pattern`.
    """
    return f"{_kebab(industry)}-{_kebab(dimension)}-pattern"


def _industry_lever_slug(industry: str, lever_kind: str) -> str:
    """Build a deterministic slug for an industry-lever pattern."""
    return f"{_kebab(industry)}-{_kebab(lever_kind)}-lever-pattern"


def _framework_rice_source_slug(industry: str, source: str) -> str:
    """Build a deterministic slug for a RICE source-weighting refinement."""
    return f"rice-{_kebab(source)}-weight-{_kebab(industry)}"


def _framework_rice_roi_slug(industry: str) -> str:
    """Build a deterministic slug for a RICE-vs-ROI divergence refinement."""
    return f"rice-roi-rank-gap-{_kebab(industry)}"


def _kebab(text: str) -> str:
    """Convert text to kebab-case slug."""
    out: list[str] = []
    prev_dash = False
    for ch in text.lower():
        if ch.isalnum():
            out.append(ch)
            prev_dash = False
        elif not prev_dash:
            out.append("-")
            prev_dash = True
    slug = "".join(out).strip("-")
    return slug or "unknown"


# ----------------------------------------------------------------------------
# Helpers
# ----------------------------------------------------------------------------


def _find_event(engagement: Engagement, kind: EventKind):  # type: ignore[no-untyped-def]
    for ev in engagement.journal.events:
        if ev.kind == kind:
            return ev
    return None


def _decimal_str_to_float(raw: object) -> float:
    if isinstance(raw, (int, float)):
        return float(raw)
    if isinstance(raw, str):
        try:
            return float(raw)
        except ValueError:
            return 0.0
    return 0.0


def _rank_key(payload: dict[str, object]) -> float:
    """Sort key — extract a numeric `rank` from a journal payload."""
    rank = payload.get("rank", 999)
    if isinstance(rank, (int, float, str)):
        try:
            return float(rank)
        except (TypeError, ValueError):
            return 999.0
    return 999.0


def _event_id_or_none(payload: dict[str, object]) -> str | None:
    """Return the `_event_id` from a payload only if it's a non-empty string."""
    eid = payload.get("_event_id")
    if isinstance(eid, str) and eid:
        return eid
    return None


__all__ = [
    "EntityType",
    "ExtractionCandidate",
    "extract_framework_refinements",
    "extract_industry_patterns",
]

"""Extraction pipeline — engagement journal → candidates → bookkeeping P8 → queue.

The end-to-end flow:

  1. `extract_industry_patterns(engagement)` + `extract_framework_refinements(engagement)`
     produce a list of `ExtractionCandidate` objects, each pre-anonymized
     via `EngagementAnonymizer`.
  2. Each candidate is wrapped as a bookkeeping `RawItem` and passed to
     `bookkeeping.score_item()` — the SAME 9-point Nous gate the rest of
     the workspace uses (no duplicate scoring math).
  3. Candidates that score ≥5/9 (`bookkeeping.PROMOTE_THRESHOLD`) land in
     the review queue at `~/.config/phronesis/extraction-queue/` as JSON
     records. They do NOT go directly to `research/entities/` — every
     candidate must clear a human review pass first (anti-pattern in the
     handoff: "don't file entities directly").
  4. Candidates that score <5 are also persisted in the queue (under
     `low-score/`) for forensic visibility — never silently dropped.

The reflexive trigger lives in `on_engagement_concluded()` — called by the
engagement model whenever an `ENGAGEMENT_CONCLUDED` event is emitted. This
makes the flow Pillar-1 (recursive self-improvement) automatic: every
completed engagement contributes to the knowledge graph without manual
invocation.

Test isolation: the queue + entity-graph paths read from env vars
`PHRONESIS_EXTRACTION_QUEUE_ROOT` and `PHRONESIS_ENTITY_GRAPH_ROOT`, both
defaulting to the operator's home. Tests set them to `tmp_path` so the
suite never touches the real knowledge graph.
"""

from __future__ import annotations

import json
import os
import sys
from collections.abc import Iterable
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

from pydantic import BaseModel, Field

from core.anonymize import AnonymizationPolicy
from core.engagement import Engagement
from core.extraction.anonymizer import (
    EngagementAnonymizer,
    _collect_engagement_redact_terms,
)
from core.extraction.candidates import (
    ExtractionCandidate,
    extract_framework_refinements,
    extract_industry_patterns,
)

# ----------------------------------------------------------------------------
# Configuration — sandboxable via env vars for test isolation.
# ----------------------------------------------------------------------------


def _resolve_queue_root() -> Path:
    """Resolve the review-queue root.

    Override via PHRONESIS_EXTRACTION_QUEUE_ROOT. Defaults to
    `~/.config/phronesis/extraction-queue/`.
    """
    env = os.environ.get("PHRONESIS_EXTRACTION_QUEUE_ROOT")
    if env:
        return Path(env).expanduser()
    return Path.home() / ".config" / "phronesis" / "extraction-queue"


def _resolve_entity_graph_root() -> Path:
    """Resolve the entity graph root.

    Override via PHRONESIS_ENTITY_GRAPH_ROOT. Defaults to
    `~/broomva/research/entities/` — the workspace knowledge graph.
    """
    env = os.environ.get("PHRONESIS_ENTITY_GRAPH_ROOT")
    if env:
        return Path(env).expanduser()
    return Path.home() / "broomva" / "research" / "entities"


# Module-level constants for backward compatibility. Tests should call the
# resolver functions directly when they need fresh env-var reads.
EXTRACTION_QUEUE_ROOT = _resolve_queue_root()
ENTITY_GRAPH_ROOT = _resolve_entity_graph_root()


# ----------------------------------------------------------------------------
# Bookkeeping interop — optional import, deterministic fallback.
# ----------------------------------------------------------------------------


def _bookkeeping_module() -> Any | None:
    """Try to import the bookkeeping module.

    Returns `None` if the module isn't installed (test environments,
    minimal CI runners). When None, `_score_candidate()` falls back to a
    deterministic stub heuristic — fine for tests, never used in
    production where the canonical bookkeeping is on PATH.
    """
    bookkeeping_path = Path.home() / "broomva" / "skills" / "bookkeeping" / "scripts"
    if not bookkeeping_path.exists():
        return None
    if str(bookkeeping_path) not in sys.path:
        sys.path.insert(0, str(bookkeeping_path))
    try:
        import bookkeeping  # type: ignore[import-not-found]

        return bookkeeping
    except ImportError:
        return None


class _CandidateScore(BaseModel):
    """Score record attached to a candidate before queue persistence."""

    total: int
    novelty: int
    specificity: int
    relevance: int
    promote: bool
    scoring_method: str
    reasoning: dict[str, Any] = Field(default_factory=dict)


def _stub_score(candidate: ExtractionCandidate) -> _CandidateScore:
    """Deterministic fallback scorer for environments without bookkeeping.

    Conservative — promotes when the candidate has rich signals (≥3 signal
    keys) AND non-empty quote AND content > 200 chars. Otherwise low score.

    Intentionally simple — the real bookkeeping P8 path is the source of
    truth in production. This stub keeps the pipeline + tests independent
    of LLM availability.
    """
    novelty = 2 if candidate.entity_type == "framework-refinement" else 1
    specificity = 2 if candidate.signals else 1
    relevance = 2 if candidate.industry or candidate.framework_ref else 1
    total = novelty + specificity + relevance
    if len(candidate.content) > 200 and candidate.quote:
        total += 1
    if len(candidate.signals) >= 3:
        total += 1

    return _CandidateScore(
        total=total,
        novelty=novelty,
        specificity=specificity,
        relevance=relevance,
        promote=total >= 5,
        scoring_method="stub-deterministic",
        reasoning={
            "rule": "stub-fallback when bookkeeping module unavailable",
        },
    )


def _score_candidate(candidate: ExtractionCandidate) -> _CandidateScore:
    """Score an extraction candidate via bookkeeping P8 (Nous gate).

    Tries the real bookkeeping module first; falls back to the
    deterministic stub if bookkeeping isn't importable. Honors the
    PHRONESIS_EXTRACTION_STUB_SCORER env var (set to "1" by tests to
    force the deterministic path).
    """
    if os.environ.get("PHRONESIS_EXTRACTION_STUB_SCORER") == "1":
        return _stub_score(candidate)

    bk = _bookkeeping_module()
    if bk is None:
        return _stub_score(candidate)

    timestamp = datetime.now(UTC).isoformat()
    raw_item = bk.RawItem(
        item_id=f"phronesis-{candidate.slug}-{timestamp}",
        source_id=f"phronesis-extraction:{candidate.slug}",
        source_type="phronesis-engagement",
        content=candidate.content,
        quote=candidate.quote,
        author="phronesis-extraction-pipeline",
        timestamp=timestamp,
        metadata={
            "entity_type": candidate.entity_type,
            "industry": candidate.industry,
            "framework_ref": candidate.framework_ref,
            "signals": candidate.signals,
            "provenance_event_ids": candidate.provenance_event_ids,
        },
    )

    try:
        existing_slugs = bk.existing_entity_slugs()
    except Exception:
        existing_slugs = []

    try:
        scored = bk.score_item(raw_item, existing_slugs)
    except Exception:
        # Network / API failure → stub fallback. Never block on transient
        # LLM failure; the candidate still lands in the review queue.
        return _stub_score(candidate)

    return _CandidateScore(
        total=int(scored.total),
        novelty=int(scored.novelty),
        specificity=int(scored.specificity),
        relevance=int(scored.relevance),
        promote=bool(scored.promote),
        scoring_method=str(scored.scoring_method),
        reasoning=dict(scored.reasoning) if isinstance(scored.reasoning, dict) else {},
    )


# ----------------------------------------------------------------------------
# Pipeline result
# ----------------------------------------------------------------------------


class ExtractionResult(BaseModel):
    """Result of an extraction pipeline run."""

    engagement_slug: str
    industry_pattern_candidates: int = 0
    framework_refinement_candidates: int = 0
    promoted_count: int = 0
    queued_count: int = 0
    queue_paths: list[Path] = Field(default_factory=list)
    promotion_paths: list[Path] = Field(default_factory=list)
    leaks: list[tuple[str, list[str]]] = Field(default_factory=list)

    @property
    def total_candidates(self) -> int:
        return self.industry_pattern_candidates + self.framework_refinement_candidates


# ----------------------------------------------------------------------------
# Public entry points
# ----------------------------------------------------------------------------


def extract_and_queue(
    engagement: Engagement,
    *,
    queue_root: Path | None = None,
    entity_graph_root: Path | None = None,
    policy: AnonymizationPolicy | None = None,
    extra_redact_terms: Iterable[str] = (),
) -> ExtractionResult:
    """Run the extraction pipeline against `engagement`.

    Args:
        engagement: a concluded engagement (state.is_concluded == True).
            Non-concluded engagements still extract but the result is
            best-effort; calling code should gate on `is_concluded` if it
            cares.
        queue_root: review-queue directory; defaults to env-resolved.
        entity_graph_root: where promoted entities land; defaults to env-resolved.
        policy: anonymization policy; defaults to strict.
        extra_redact_terms: caller-supplied redact terms (project codenames
            the journal can't surface automatically).

    Returns:
        `ExtractionResult` summarizing what was queued + promoted + leaked.

    Side effects:
        Writes JSON queue records to `<queue_root>/<engagement_slug>/`.
        Writes promoted candidate stubs to
        `<entity_graph_root>/<entity_type>/<slug>.md` IF the bookkeeping
        score is ≥5 AND no canary leaks are detected post-redaction.
        The promoted file is a STUB — operator polishes before
        publishing. We never short-circuit the human review pass.
    """
    queue_root = queue_root or _resolve_queue_root()
    entity_graph_root = entity_graph_root or _resolve_entity_graph_root()
    policy = policy or AnonymizationPolicy()

    journal_terms = _collect_engagement_redact_terms(engagement)
    combined_terms = list(
        dict.fromkeys([*policy.redact_terms, *journal_terms, *extra_redact_terms])
    )
    effective_policy = policy.model_copy(update={"redact_terms": combined_terms})

    anonymizer = EngagementAnonymizer(tenant=engagement.tenant, policy=effective_policy)

    industry_candidates = extract_industry_patterns(engagement, anonymizer)
    framework_candidates = extract_framework_refinements(engagement, anonymizer)
    all_candidates = [*industry_candidates, *framework_candidates]

    result = ExtractionResult(
        engagement_slug=engagement.tenant.tenant_slug,
        industry_pattern_candidates=len(industry_candidates),
        framework_refinement_candidates=len(framework_candidates),
    )

    queue_dir = queue_root / engagement.tenant.tenant_slug
    queue_dir.mkdir(parents=True, exist_ok=True)

    timestamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")

    for candidate in all_candidates:
        # Defense-in-depth: re-check for tenant markers AFTER candidate
        # construction (in case a custom anonymizer wrapper was passed and
        # didn't strip something). Never queue a candidate that carries a
        # tenant marker — the canary release gate would catch it later
        # but we'd rather fail fast.
        leaked = anonymizer.carries_marker(candidate.content) + anonymizer.carries_marker(
            candidate.quote
        )
        if leaked:
            result.leaks.append((candidate.slug, leaked))
            continue

        score = _score_candidate(candidate)
        promoted = score.promote

        record = {
            "candidate": candidate.model_dump(mode="json"),
            "score": score.model_dump(mode="json"),
            "engagement_slug": engagement.tenant.tenant_slug,
            "promoted_at": datetime.now(UTC).isoformat(),
        }

        sub_dir = "promoted" if promoted else "low-score"
        out_dir = queue_dir / sub_dir
        out_dir.mkdir(parents=True, exist_ok=True)
        out_path = out_dir / f"{candidate.slug}-{timestamp}.json"
        out_path.write_text(json.dumps(record, indent=2, sort_keys=False, default=str))
        result.queue_paths.append(out_path)

        if promoted:
            result.promoted_count += 1
            entity_dir = entity_graph_root / candidate.entity_type
            entity_dir.mkdir(parents=True, exist_ok=True)
            entity_path = entity_dir / f"{candidate.slug}.md"
            entity_path.write_text(_render_entity_stub(candidate, score, engagement))
            result.promotion_paths.append(entity_path)
        else:
            result.queued_count += 1

    return result


def on_engagement_concluded(
    engagement: Engagement,
    *,
    enabled: bool | None = None,
    **kwargs: Any,
) -> ExtractionResult | None:
    """Reflexive hook — called when `ENGAGEMENT_CONCLUDED` fires.

    Per `feedback_bookkeeping_reflexive.md`: bookkeeping is reflexive,
    not opt-in. This hook runs without being asked.

    Set `PHRONESIS_EXTRACTION_ENABLED=0` to disable (e.g. during
    interactive debugging of a non-concluded fixture). Default: enabled.

    Returns the `ExtractionResult`, or `None` if disabled.
    """
    if enabled is None:
        enabled = os.environ.get("PHRONESIS_EXTRACTION_ENABLED", "1") != "0"
    if not enabled:
        return None
    if not engagement.state().is_concluded:
        # Best-effort idempotency: only fire on truly-concluded engagements.
        return None
    return extract_and_queue(engagement, **kwargs)


# ----------------------------------------------------------------------------
# Entity page stub renderer
# ----------------------------------------------------------------------------


def _render_entity_stub(
    candidate: ExtractionCandidate,
    score: _CandidateScore,
    engagement: Engagement,
) -> str:
    """Render a minimal entity-page stub for a promoted candidate.

    Format mirrors `~/broomva/skills/bookkeeping/templates/entity-page.md`
    minimally — just enough for the file to be lint-clean. Operator
    polishes before the entity surfaces in queries.

    Stays in YAML frontmatter + plain markdown body. Per the workspace
    Format Discernment rule, entity pages are Category A (substrate),
    so this is markdown-only — never HTML.
    """
    industry_or_framework = (
        f"  industry: {candidate.industry}"
        if candidate.industry
        else f"  framework_ref: {candidate.framework_ref}"
    )
    signals_lines = "\n".join(f"  - {k}: {v}" for k, v in candidate.signals.items())
    body = candidate.content

    return f"""---
type: {candidate.entity_type}
slug: {candidate.slug}
title: {candidate.title}
status: candidate
provenance:
  source: phronesis-extraction
  engagement_slug: {engagement.tenant.tenant_slug}
  event_ids:
{chr(10).join(f"    - {eid}" for eid in candidate.provenance_event_ids)}
score:
  total: {score.total}/9
  novelty: {score.novelty}
  specificity: {score.specificity}
  relevance: {score.relevance}
  method: {score.scoring_method}
{industry_or_framework}
signals:
{signals_lines}
created_at: {datetime.now(UTC).isoformat()}
---

# {candidate.title}

## Pattern (anonymized)

{body}

## Source quote

> {candidate.quote}

## Promotion gate

- Bookkeeping P8 score: {score.total}/9 ({score.scoring_method})
- Rule-of-three: this is one instance. Surface ≥2 more same-industry /
  same-framework engagements reproducing the pattern before treating as
  a stable graph node.
- Operator action: polish the body, verify the signals, confirm the
  anonymization holds, then promote `status` from `candidate` to `active`.
"""


__all__ = [
    "EXTRACTION_QUEUE_ROOT",
    "ENTITY_GRAPH_ROOT",
    "ExtractionResult",
    "extract_and_queue",
    "on_engagement_concluded",
]

"""Framework registry — loads all YAMLs at frameworks/ and validates them
against frameworks/_schema.yaml.

Phase 1 ships 14 implemented frameworks + 13 D-scope stubs.
Phase 3 mirrors this in life-phronesis Rust crate.

P7 enforcement: framework_selector caps active frameworks at ≤5 per
engagement. The registry just loads + validates; selection happens in
core/selector.py.
"""

from __future__ import annotations

from pathlib import Path
from typing import Any

import yaml
from pydantic import BaseModel, Field

FRAMEWORKS_ROOT = Path(__file__).resolve().parent.parent / "frameworks"


class FrameworkInput(BaseModel):
    """One input parameter a framework consumes."""

    name: str
    type: str
    description: str


class Framework(BaseModel):
    """In-memory representation of a framework YAML.

    Mirrors frameworks/_schema.yaml. Pydantic does the runtime validation;
    the YAML schema gets validated separately via jsonschema in
    `load_framework` so we get JSON-Schema-quality error messages.

    The YAML field `when_NOT_to_use` (mixed case per design spec Appendix A)
    is exposed as `when_not_to_use` in Python via Pydantic alias. Both names
    work on construction — population_by_field_name is enabled.
    """

    model_config = {"populate_by_name": True}

    id: str
    name: str
    source_firm: str
    source_year: int
    source_citation: str
    category: str
    purpose: str
    inputs: list[FrameworkInput] = Field(default_factory=list)
    dimensions: list[str] = Field(default_factory=list)
    scoring_rubric: dict[str, Any]
    output_shape: dict[str, Any]
    when_to_use: list[str]
    when_not_to_use: list[str] = Field(alias="when_NOT_to_use")
    example_application: dict[str, Any] = Field(default_factory=dict)
    relationships: dict[str, list[str]] = Field(default_factory=dict)
    citations: list[str] = Field(default_factory=list)
    is_d_scope: bool = False


def _validate_against_schema(raw: dict[str, Any]) -> None:
    """Validate a YAML dict against frameworks/_schema.yaml using jsonschema.

    Raises jsonschema.ValidationError on first violation. We keep this
    optional — if jsonschema isn't installed the Pydantic layer still
    catches missing/wrong-typed fields.
    """
    schema_path = FRAMEWORKS_ROOT / "_schema.yaml"
    if not schema_path.exists():
        return
    try:
        import jsonschema
    except ImportError:
        return
    schema = yaml.safe_load(schema_path.read_text())
    jsonschema.validate(raw, schema)


def load_framework(path: Path) -> Framework:
    """Load and validate one framework YAML.

    Validates against `_schema.yaml` first (JSON Schema), then through
    Pydantic. Two layers because they catch different classes of error:
    JSON Schema catches structural issues with rich error messages;
    Pydantic catches type-narrowing issues and gives clean Python objects.
    """
    raw = yaml.safe_load(path.read_text())
    if not isinstance(raw, dict):
        raise ValueError(f"Framework YAML must be a top-level mapping; got {type(raw).__name__}")
    _validate_against_schema(raw)
    return Framework.model_validate(raw)


def load_all() -> dict[str, Framework]:
    """Load every framework YAML under FRAMEWORKS_ROOT, keyed by id.

    Skips `_schema.yaml`. Raises if two frameworks share an id, or if any
    YAML fails validation.
    """
    out: dict[str, Framework] = {}
    if not FRAMEWORKS_ROOT.exists():
        return out
    for yaml_path in sorted(FRAMEWORKS_ROOT.rglob("*.yaml")):
        if yaml_path.name == "_schema.yaml":
            continue
        fw = load_framework(yaml_path)
        if fw.id in out:
            raise ValueError(f"Duplicate framework id: {fw.id!r} (at {yaml_path} and earlier)")
        out[fw.id] = fw
    return out

"""Phronesis linter — L1-L5 + P3 + P7 + P8 rules over a rendered engagement.

Each rule scans the journal (and optionally the rendered deliverables) and
emits LintViolation objects. The release-gate test (Phase E) requires zero
L1-L5 errors on the canonical synthetic fixture.

Stages already enforce gates at their boundaries (raise ValueError when
the gate is violated). The linter is the BACKSTOP — replaying a finished
engagement journal must produce zero L-errors. If a linter rule fires, it
means a stage was bypassed or the engagement was constructed manually
without going through the stages.
"""

from __future__ import annotations

from collections import Counter
from collections.abc import Callable

from pydantic import BaseModel, Field

from core.engagement import Engagement
from core.types import EventKind


class LintViolation(BaseModel):
    """One linter finding."""

    rule: str  # "L1" | "L2" | "L3" | "L4" | "L5" | "P3" | "P7" | "P8"
    severity: str  # "error" | "warning"
    location: str  # event_id, deliverable slug, or descriptive marker
    message: str


class LintResult(BaseModel):
    """Aggregated result of running all linter rules over an Engagement."""

    violations: list[LintViolation] = Field(default_factory=list)

    @property
    def has_errors(self) -> bool:
        return any(v.severity == "error" for v in self.violations)

    @property
    def has_warnings(self) -> bool:
        return any(v.severity == "warning" for v in self.violations)

    def errors_for_rule(self, rule: str) -> list[LintViolation]:
        return [v for v in self.violations if v.rule == rule and v.severity == "error"]


# Rules registry ------------------------------------------------------------

_RULES: list[Callable[[Engagement], list[LintViolation]]] = []


def rule(
    fn: Callable[[Engagement], list[LintViolation]],
) -> Callable[[Engagement], list[LintViolation]]:
    """Decorator: register a linter rule."""
    _RULES.append(fn)
    return fn


# L1 — STRATEGIC_THESIS_REQUIRED ---------------------------------------------


@rule
def l1_strategic_thesis_required(eng: Engagement) -> list[LintViolation]:
    """L1 — every engagement must declare a StrategicThesis BEFORE leaving Stage 1.

    Bision Failure 1 (100% observed): Sin tesis estratégica.
    Engagements that emitted INTAKE_COMPLETED without first emitting
    STRATEGIC_THESIS_DECLARED are blocked.
    """
    violations: list[LintViolation] = []
    state = eng.state()

    intake_closed_event_id: str | None = None
    for ev in eng.journal.events:
        if ev.kind == EventKind.INTAKE_COMPLETED:
            intake_closed_event_id = ev.event_id
            break

    if intake_closed_event_id and state.thesis_id is None:
        violations.append(
            LintViolation(
                rule="L1",
                severity="error",
                location=intake_closed_event_id,
                message=(
                    "L1 STRATEGIC_THESIS_REQUIRED — intake closed without a "
                    "thesis declared. Bision Failure 1 (100% observed): "
                    "'sin tesis estratégica'."
                ),
            )
        )
    return violations


# L2 — DIVERSE_IDEATION_SOURCES ----------------------------------------------


@rule
def l2_diverse_ideation_sources(eng: Engagement) -> list[LintViolation]:
    """L2 — ideation diversity rule.

    Bision Failure 2 (87% observed): casos mal priorizados — elección por novedad.
    Triggers when ≥1 USE_CASE_PROPOSED exists AND either:
      - <3 distinct ideation sources OR
      - NOVELTY share > 50%
    """
    violations: list[LintViolation] = []
    state = eng.state()
    if not state.use_cases:
        return violations

    sources = [uc.get("ideation_source") for uc in state.use_cases.values()]
    sources_counter = Counter(s for s in sources if isinstance(s, str))
    distinct = len(sources_counter)
    novelty_share = sources_counter.get("novelty", 0) / len(state.use_cases)

    if distinct < 3:
        violations.append(
            LintViolation(
                rule="L2",
                severity="error",
                location="ideate-stage",
                message=(
                    f"L2 DIVERSE_IDEATION_SOURCES — only {distinct} distinct "
                    f"ideation sources across {len(state.use_cases)} use cases "
                    f"(min 3). Bision Failure 2 (87% observed)."
                ),
            )
        )
    if novelty_share > 0.5:
        violations.append(
            LintViolation(
                rule="L2",
                severity="error",
                location="ideate-stage",
                message=(
                    f"L2 DIVERSE_IDEATION_SOURCES — NOVELTY share "
                    f"{novelty_share:.0%} exceeds 50% limit. "
                    f"Bision Failure 2: elección por novedad."
                ),
            )
        )
    return violations


# L3 — DATA_READINESS_GATE ---------------------------------------------------


@rule
def l3_data_readiness_gate(eng: Engagement) -> list[LintViolation]:
    """L3 — data readiness must be assessed for every proposed use case.

    Bision Failure 3 (74% observed): datos no preparados.
    Triggers when a use case has `readiness_band == "blocking"` AND no
    `prep_phase_required` AND the engagement is past Stage 3.
    """
    violations: list[LintViolation] = []
    state = eng.state()
    if not state.use_cases:
        return violations

    for uc_id, payload in state.use_cases.items():
        band = payload.get("data_readiness_band")
        if band == "blocking":
            violations.append(
                LintViolation(
                    rule="L3",
                    severity="error",
                    location=f"use-case:{uc_id}",
                    message=(
                        f"L3 DATA_READINESS_GATE — use case {uc_id!r} has "
                        f"data_readiness_band=blocking. Bision Failure 3 "
                        f"(74% observed): datos no preparados. Either declare "
                        f"a prep_phase or defer the use case."
                    ),
                )
            )
    return violations


# L4 — ADOPTION_METRIC_REQUIRED ----------------------------------------------


@rule
def l4_adoption_metric_required(eng: Engagement) -> list[LintViolation]:
    """L4 — every PILOT_STARTED must reference a pilot design with an
    adoption_metric. Pydantic enforces non-emptiness on PilotDesign at
    construction; this lints the journal sequence.

    Bision Failure 4 (61% observed): desconexión negocio-tecnología —
    modelos con precisión alta y baja adopción.

    Phase 1 backstop: PilotDesign requires AdoptionMetric (typed). The linter
    surfaces L4 if a PILOT_STARTED event is followed by no DELIVERABLE_RENDERED
    for slug "pilot-plan" — i.e., the engagement skipped pilot-plan rendering,
    so the adoption metric never reached the deliverable layer.
    """
    violations: list[LintViolation] = []
    pilots: list[str] = []
    rendered_pilot_plans: list[str] = []
    for ev in eng.journal.events:
        if ev.kind == EventKind.PILOT_STARTED:
            uc_id = ev.payload.get("use_case_id")
            if isinstance(uc_id, str):
                pilots.append(uc_id)
        elif ev.kind == EventKind.DELIVERABLE_RENDERED:
            slug = ev.payload.get("slug")
            if slug == "pilot-plan":
                rendered_pilot_plans.append(ev.event_id)

    if pilots and not rendered_pilot_plans:
        violations.append(
            LintViolation(
                rule="L4",
                severity="error",
                location="roadmap-stage",
                message=(
                    "L4 ADOPTION_METRIC_REQUIRED — pilot(s) started "
                    f"({pilots}) but no pilot-plan deliverable rendered. "
                    "Bision Failure 4 (61% observed): adoption metric must "
                    "land in the deliverable so it's audit-trail visible."
                ),
            )
        )
    return violations


# L5 — BASELINE_REQUIRED -----------------------------------------------------


@rule
def l5_baseline_required(eng: Engagement) -> list[LintViolation]:
    """L5 — every PILOT_STARTED must be preceded by ≥1 BASELINE_CAPTURED.

    Bision Failure 5 (48% observed): sin medición de ROI — no retroactive
    baselines. RoadmapStage.design_pilot enforces this at construction; the
    linter is the journal-replay backstop.
    """
    violations: list[LintViolation] = []
    baseline_count_before: dict[str, int] = {}
    captured_so_far = 0
    for ev in eng.journal.events:
        if ev.kind == EventKind.BASELINE_CAPTURED:
            captured_so_far += 1
        elif ev.kind == EventKind.PILOT_STARTED:
            uc_id = ev.payload.get("use_case_id")
            if isinstance(uc_id, str):
                baseline_count_before[uc_id] = captured_so_far

    for uc_id, count in baseline_count_before.items():
        if count == 0:
            violations.append(
                LintViolation(
                    rule="L5",
                    severity="error",
                    location=f"pilot:{uc_id}",
                    message=(
                        f"L5 BASELINE_REQUIRED — pilot for {uc_id!r} started "
                        f"with zero prior BASELINE_CAPTURED events. "
                        f"Bision Failure 5 (48% observed): no retroactive "
                        f"baselines. Capture before pilot."
                    ),
                )
            )
    return violations


# Aggregate runner ----------------------------------------------------------


def lint_engagement(eng: Engagement) -> LintResult:
    """Run all registered rules; return aggregated result."""
    out = LintResult()
    for fn in _RULES:
        out.violations.extend(fn(eng))
    return out