omi/backend/models/fair_use.py at main · ghzhost/omi · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
from datetime import datetime, timezone
from enum import Enum
from typing import List, Optional

from pydantic import BaseModel, Field


def _utc_now() -> datetime:
    return datetime.now(timezone.utc).replace(tzinfo=None)


class FairUseStage(str, Enum):
    """Graduated enforcement stages."""

    NONE = "none"
    WARNING = "warning"
    THROTTLE = "throttle"
    RESTRICT = "restrict"


class UsageType(str, Enum):
    """Types of detected non-personal usage."""

    NONE = "none"
    AUDIOBOOK = "audiobook"
    PODCAST = "podcast"
    PRERECORDED = "prerecorded"
    TV_MOVIE = "tv_movie"
    COMMERCIAL = "commercial"
    UNKNOWN = "unknown"
    FREE_EXHAUSTED = "free_exhausted"


class SoftCapTrigger(str, Enum):
    """Which rolling window triggered the soft cap."""

    DAILY = "daily"
    THREE_DAY = "3day"
    WEEKLY = "weekly"


class ClassifierEvidence(BaseModel):
    """Evidence from a single conversation flagged by the LLM classifier."""

    conversation_id: str
    title: str = ""
    category: str = ""
    reason: str = ""


class ClassifierResult(BaseModel):
    """Result from the LLM fair-use classifier."""

    model: str = ""
    prompt_version: str = "v2"
    misuse_score: float = 0.0
    usage_type: UsageType = UsageType.NONE
    confidence: float = 0.0
    evidence: List[ClassifierEvidence] = Field(default_factory=list)


class FairUseState(BaseModel):
    """Per-user fair use enforcement state. Stored at users/{uid}/fair_use_state/current."""

    stage: FairUseStage = FairUseStage.NONE
    violation_count_7d: int = 0
    violation_count_30d: int = 0
    last_violation_at: Optional[datetime] = None
    throttle_until: Optional[datetime] = None
    restrict_until: Optional[datetime] = None
    last_classifier_score: float = 0.0
    last_classifier_type: UsageType = UsageType.NONE
    updated_at: datetime = Field(default_factory=_utc_now)


class FairUseEvent(BaseModel):
    """A single fair-use violation event. Stored at users/{uid}/fair_use_events/{event_id}."""

    created_at: datetime = Field(default_factory=_utc_now)
    session_id: str = ""
    trigger: SoftCapTrigger = SoftCapTrigger.DAILY
    window_speech_ms: dict[str, int] = Field(default_factory=dict[str, int])  # {daily, three_day, weekly}
    thresholds_ms: dict[str, int] = Field(default_factory=dict[str, int])  # snapshot of active thresholds
    classifier: Optional[ClassifierResult] = None
    enforcement_action: str = ""  # warning, throttle, restrict, none
    previous_stage: FairUseStage = FairUseStage.NONE
    new_stage: FairUseStage = FairUseStage.NONE
    admin_notes: str = ""
    resolved: bool = False
    resolved_at: Optional[datetime] = None
    resolved_by: str = ""


class FairUseUserSummary(BaseModel):
    """Summary for admin dashboard."""

    uid: str
    stage: FairUseStage = FairUseStage.NONE
    violation_count_7d: int = 0
    violation_count_30d: int = 0
    last_violation_at: Optional[datetime] = None
    last_classifier_score: float = 0.0
    last_classifier_type: UsageType = UsageType.NONE
    speech_hours_today: float = 0.0
    speech_hours_7d: float = 0.0