|
| 1 | +""" |
| 2 | +coverage_state.py |
| 3 | +───────────────── |
| 4 | +Immutable coverage state model for the Topic Coverage Tracker. |
| 5 | +
|
| 6 | +Responsibilities |
| 7 | +──────────────── |
| 8 | +- Represent the full coverage snapshot at any point in the viva. |
| 9 | +- Track per-topic entry records (asked count, tags, outcomes). |
| 10 | +- Support functional updates (replace, not mutate). |
| 11 | +- Be trivially serializable (plain Python types only). |
| 12 | +
|
| 13 | +Rules |
| 14 | +───── |
| 15 | +- Frozen dataclasses — state is replaced, never mutated. |
| 16 | +- All update methods return a new CoverageState. |
| 17 | +- No dependency on ORACLE internals. |
| 18 | +- Serialization: model_dump() → plain dict with no custom types. |
| 19 | +""" |
| 20 | + |
| 21 | +from __future__ import annotations |
| 22 | + |
| 23 | +from dataclasses import dataclass, field, replace |
| 24 | +from enum import Enum |
| 25 | +from typing import Dict, FrozenSet, List, Optional, Tuple |
| 26 | + |
| 27 | + |
| 28 | +# ── Coverage status ─────────────────────────────────────────────────────────── |
| 29 | + |
| 30 | +class CoverageStatus(str, Enum): |
| 31 | + NOT_STARTED = "not_started" # Category has 0 questions asked |
| 32 | + PARTIAL = "partial" # Below min_questions threshold |
| 33 | + COVERED = "covered" # Met or exceeded min_questions |
| 34 | + SATURATED = "saturated" # Significantly over min_questions (diminishing returns) |
| 35 | + |
| 36 | + |
| 37 | +# ── Topic entry ─────────────────────────────────────────────────────────────── |
| 38 | + |
| 39 | +@dataclass(frozen=True) |
| 40 | +class TopicEntry: |
| 41 | + """ |
| 42 | + Immutable record of all asks for a single question_target topic. |
| 43 | +
|
| 44 | + question_target : Canonical topic identifier (matches NormalizedVivaTarget). |
| 45 | + category : Which coverage domain this belongs to. |
| 46 | + tags : Lowercase topic tags for this entry. |
| 47 | + ask_count : How many times this topic has been asked. |
| 48 | + answered : True if at least one answer was received. |
| 49 | + turn_first_asked: Turn index when first asked (for ordering). |
| 50 | + """ |
| 51 | + question_target: str |
| 52 | + category: str |
| 53 | + tags: FrozenSet[str] |
| 54 | + ask_count: int = 0 |
| 55 | + answered: bool = False |
| 56 | + turn_first_asked: int = 0 |
| 57 | + |
| 58 | + def increment(self, answered: bool = False) -> "TopicEntry": |
| 59 | + """Return a new entry with ask_count+1 and optional answered flag.""" |
| 60 | + return replace( |
| 61 | + self, |
| 62 | + ask_count = self.ask_count + 1, |
| 63 | + answered = self.answered or answered, |
| 64 | + ) |
| 65 | + |
| 66 | + def to_dict(self) -> dict: |
| 67 | + return { |
| 68 | + "question_target": self.question_target, |
| 69 | + "category": self.category, |
| 70 | + "tags": sorted(self.tags), |
| 71 | + "ask_count": self.ask_count, |
| 72 | + "answered": self.answered, |
| 73 | + "turn_first_asked": self.turn_first_asked, |
| 74 | + } |
| 75 | + |
| 76 | + |
| 77 | +# ── Coverage state ──────────────────────────────────────────────────────────── |
| 78 | + |
| 79 | +@dataclass(frozen=True) |
| 80 | +class CoverageState: |
| 81 | + """ |
| 82 | + Complete, immutable snapshot of topic coverage at one point in the viva. |
| 83 | +
|
| 84 | + topics : Dict[question_target → TopicEntry] |
| 85 | + category_counts : Dict[category_name → total ask_count] |
| 86 | + covered_categories: Set of category names that met their min_questions. |
| 87 | + total_turns : Total viva turns elapsed (for saturation checks). |
| 88 | + """ |
| 89 | + topics: Dict[str, TopicEntry] = field(default_factory=dict) |
| 90 | + category_counts: Dict[str, int] = field(default_factory=dict) |
| 91 | + covered_categories: FrozenSet[str] = field(default_factory=frozenset) |
| 92 | + total_turns: int = 0 |
| 93 | + |
| 94 | + # ── Derived properties ──────────────────────────────────────────────────── |
| 95 | + |
| 96 | + @property |
| 97 | + def asked_targets(self) -> FrozenSet[str]: |
| 98 | + return frozenset(self.topics.keys()) |
| 99 | + |
| 100 | + @property |
| 101 | + def answered_targets(self) -> FrozenSet[str]: |
| 102 | + return frozenset(t for t, e in self.topics.items() if e.answered) |
| 103 | + |
| 104 | + @property |
| 105 | + def total_asked(self) -> int: |
| 106 | + return len(self.topics) |
| 107 | + |
| 108 | + def category_ask_count(self, category: str) -> int: |
| 109 | + return self.category_counts.get(category, 0) |
| 110 | + |
| 111 | + def is_covered(self, category: str) -> bool: |
| 112 | + return category in self.covered_categories |
| 113 | + |
| 114 | + def has_been_asked(self, question_target: str) -> bool: |
| 115 | + return question_target in self.topics |
| 116 | + |
| 117 | + # ── Functional updates ──────────────────────────────────────────────────── |
| 118 | + |
| 119 | + def record_ask( |
| 120 | + self, |
| 121 | + question_target: str, |
| 122 | + category: str, |
| 123 | + tags: FrozenSet[str], |
| 124 | + turn_index: int, |
| 125 | + min_questions_for_category: int = 1, |
| 126 | + ) -> "CoverageState": |
| 127 | + """ |
| 128 | + Return new state reflecting a question being asked. |
| 129 | +
|
| 130 | + Creates a new TopicEntry if the target hasn't been seen before. |
| 131 | + """ |
| 132 | + existing = self.topics.get(question_target) |
| 133 | + if existing: |
| 134 | + updated_entry = existing.increment(answered=False) |
| 135 | + else: |
| 136 | + updated_entry = TopicEntry( |
| 137 | + question_target = question_target, |
| 138 | + category = category, |
| 139 | + tags = tags, |
| 140 | + ask_count = 1, |
| 141 | + answered = False, |
| 142 | + turn_first_asked = turn_index, |
| 143 | + ) |
| 144 | + |
| 145 | + new_topics = {**self.topics, question_target: updated_entry} |
| 146 | + new_counts = dict(self.category_counts) |
| 147 | + new_counts[category] = new_counts.get(category, 0) + 1 |
| 148 | + |
| 149 | + # Recompute covered categories |
| 150 | + covered = set(self.covered_categories) |
| 151 | + if new_counts[category] >= min_questions_for_category: |
| 152 | + covered.add(category) |
| 153 | + |
| 154 | + return replace( |
| 155 | + self, |
| 156 | + topics = new_topics, |
| 157 | + category_counts = new_counts, |
| 158 | + covered_categories = frozenset(covered), |
| 159 | + total_turns = self.total_turns + 1, |
| 160 | + ) |
| 161 | + |
| 162 | + def record_answer(self, question_target: str) -> "CoverageState": |
| 163 | + """Return new state marking the topic as answered.""" |
| 164 | + if question_target not in self.topics: |
| 165 | + return self |
| 166 | + updated = self.topics[question_target].increment(answered=True) |
| 167 | + new_topics = {**self.topics, question_target: updated} |
| 168 | + return replace(self, topics=new_topics) |
| 169 | + |
| 170 | + def advance_turn(self) -> "CoverageState": |
| 171 | + """Increment total_turns without recording a topic.""" |
| 172 | + return replace(self, total_turns=self.total_turns + 1) |
| 173 | + |
| 174 | + # ── Serialization ───────────────────────────────────────────────────────── |
| 175 | + |
| 176 | + def to_dict(self) -> dict: |
| 177 | + """Return a plain-dict representation for export / logging.""" |
| 178 | + return { |
| 179 | + "topics": {k: v.to_dict() for k, v in self.topics.items()}, |
| 180 | + "category_counts": dict(self.category_counts), |
| 181 | + "covered_categories": sorted(self.covered_categories), |
| 182 | + "total_turns": self.total_turns, |
| 183 | + "total_asked": self.total_asked, |
| 184 | + } |
| 185 | + |
| 186 | + @classmethod |
| 187 | + def empty(cls) -> "CoverageState": |
| 188 | + """Return a zeroed CoverageState for session initialization.""" |
| 189 | + return cls() |
0 commit comments