agentic-resources/session_memory/curate/schema.py

"""Solution Pattern schema (PRD §6.3 FR-U2; design OQ4) — T01.

A **Solution Pattern** is the curated, reviewed artifact a candidate pattern is
promoted into: a named, versioned record pairing a problem (or success) with one
or more recommended resolutions, written **flavor-agnostically**. Everything a
distributor needs to render a native artifact lives in a *separate*
``rendering_hints`` sub-structure, keyed by flavor — so the core stays neutral
(FR-A1/FR-A2) while Phase 3 distributors still get enough to render well (OQ4).

The artifact is the durable unit of the Pattern Catalog (T02): files originate,
the State Hub indexes (ADR-001). Serialization is deterministic (sorted keys) so
catalog files diff cleanly and re-saving an unchanged pattern is a no-op.
"""

from __future__ import annotations

import json
import re
from dataclasses import asdict, dataclass, field, fields
from typing import Any, Optional

from ..core.schema import FLAVORS

SCHEMA_VERSION = 1

# Lifecycle of a catalogued pattern.
#   provisional — promoted but below the distribution evidence bar (OQ5)
#   approved    — meets the bar; distribution-eligible (Phase 3)
#   rejected    — reviewed and declined; remembered so it is not re-surfaced
#   superseded  — replaced by a newer version of the same pattern id
STATUSES = ("provisional", "approved", "rejected", "superseded")

POLARITIES = ("problem", "success")


@dataclass
class Resolution:
    """One recommended resolution for the pattern's problem (FR-U2)."""

    summary: str
    detail: str = ""
    steps: list[str] = field(default_factory=list)


@dataclass
class Scope:
    """Where the pattern applies (FR-X2 input). Empty list == unrestricted."""

    repos: list[str] = field(default_factory=list)
    domains: list[str] = field(default_factory=list)
    flavors: list[str] = field(default_factory=list)

    def __post_init__(self) -> None:
        bad = [f for f in self.flavors if f not in FLAVORS]
        if bad:
            raise ValueError(f"unknown flavor(s) in scope {bad!r}; expected {FLAVORS}")


@dataclass
class Provenance:
    """Trace back to the detect candidate this pattern was promoted from."""

    source_key: str  # the detect Pattern.key — stable cluster identity
    evidence: dict[str, Any] = field(default_factory=dict)  # snapshot of the candidate
    detected_at: Optional[str] = None
    promoted_at: Optional[str] = None


@dataclass
class SolutionPattern:
    """A curated, versioned solution pattern (PRD §5 / §6.3)."""

    id: str  # stable, derived from provenance.source_key
    name: str
    version: str  # semantic, e.g. "1.0.0"
    polarity: str  # problem | success
    problem: str  # human-readable description of the recurring situation
    resolutions: list[Resolution] = field(default_factory=list)
    scope: Scope = field(default_factory=Scope)
    provenance: Provenance = field(default_factory=lambda: Provenance(source_key=""))
    # per-flavor rendering hints, kept OUT of the agnostic core (OQ4):
    #   {"claude": {...}, "codex": {...}, "grok": {...}}
    rendering_hints: dict[str, dict[str, Any]] = field(default_factory=dict)
    status: str = "provisional"
    distribution_ready: bool = False
    created_at: Optional[str] = None
    updated_at: Optional[str] = None
    schema_version: int = SCHEMA_VERSION

    def __post_init__(self) -> None:
        if self.polarity not in POLARITIES:
            raise ValueError(f"unknown polarity {self.polarity!r}; expected {POLARITIES}")
        if self.status not in STATUSES:
            raise ValueError(f"unknown status {self.status!r}; expected {STATUSES}")
        bad = [f for f in self.rendering_hints if f not in FLAVORS]
        if bad:
            raise ValueError(f"unknown flavor(s) in rendering_hints {bad!r}; expected {FLAVORS}")

    # --- identity / versioning helpers -------------------------------------

    @staticmethod
    def make_id(source_key: str) -> str:
        """Stable catalog id from a detect candidate key (``polarity:type:locus``).

        Identity is the source key, so re-promoting the same candidate maps to the
        same pattern (dedup in T02), independent of wording or version.
        """
        slug = re.sub(r"[^a-z0-9_]+", "-", source_key.lower()).strip("-")
        return f"sp-{slug}"

    @staticmethod
    def bump_version(version: str, level: str = "patch") -> str:
        """Increment a ``major.minor.patch`` version string."""
        parts = (version.split(".") + ["0", "0", "0"])[:3]
        major, minor, patch = (int(p) for p in parts)
        if level == "major":
            major, minor, patch = major + 1, 0, 0
        elif level == "minor":
            minor, patch = minor + 1, 0
        else:
            patch += 1
        return f"{major}.{minor}.{patch}"

    # --- serialization ------------------------------------------------------

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)

    def to_json(self) -> str:
        return json.dumps(self.to_dict(), sort_keys=True, indent=2)

    @classmethod
    def from_dict(cls, d: dict[str, Any]) -> "SolutionPattern":
        d = dict(d)
        resolutions = [Resolution(**{k: v for k, v in r.items() if k in _RESOLUTION_FIELDS})
                       for r in d.pop("resolutions", [])]
        scope = d.pop("scope", None)
        prov = d.pop("provenance", None)
        obj = cls(**{k: v for k, v in d.items() if k in _PATTERN_FIELDS})
        obj.resolutions = resolutions
        if scope is not None:
            obj.scope = Scope(**{k: v for k, v in scope.items() if k in _SCOPE_FIELDS})
        if prov is not None:
            obj.provenance = Provenance(**{k: v for k, v in prov.items() if k in _PROV_FIELDS})
        return obj

    @classmethod
    def from_json(cls, s: str) -> "SolutionPattern":
        return cls.from_dict(json.loads(s))


_PATTERN_FIELDS = {f.name for f in fields(SolutionPattern)}
_RESOLUTION_FIELDS = {f.name for f in fields(Resolution)}
_SCOPE_FIELDS = {f.name for f in fields(Scope)}
_PROV_FIELDS = {f.name for f in fields(Provenance)}