escalations.py
python
sha256:e8214e0062ef8ef0af999937df2731655b6082781fc22bc563f58db2f42b1de9
Merge 'bump/muse-v0.2.1' into 'dev' — proposal: chore: bump…
Human
10 days ago
| 1 | """Harmony escalation persistence — record, load, list, resolve escalations. |
| 2 | |
| 3 | Single responsibility: CRUD for EscalationRecord objects in the harmony store. |
| 4 | Uses the raw os.replace write pattern (like the original) because escalation |
| 5 | records carry fsync for durability. |
| 6 | """ |
| 7 | |
| 8 | from __future__ import annotations |
| 9 | |
| 10 | import datetime |
| 11 | import json |
| 12 | import logging |
| 13 | import os |
| 14 | import pathlib |
| 15 | import tempfile |
| 16 | from collections.abc import Mapping |
| 17 | from dataclasses import replace as dc_replace |
| 18 | |
| 19 | from muse.core.types import JsonValue, short_id |
| 20 | |
| 21 | from .paths import ( |
| 22 | escalation_path, |
| 23 | escalations_dir, |
| 24 | _validate_id, |
| 25 | ) |
| 26 | from .types import ( |
| 27 | AgentProvenance, |
| 28 | EscalationRecord, |
| 29 | EscalationStatus, |
| 30 | _EscalationDict, |
| 31 | ) |
| 32 | |
| 33 | logger = logging.getLogger(__name__) |
| 34 | |
| 35 | #: Maximum bytes read from a single escalation record file. |
| 36 | _MAX_ESCALATION_BYTES: int = 16_384 # 16 KiB |
| 37 | |
| 38 | # --------------------------------------------------------------------------- |
| 39 | # Serialisation helpers |
| 40 | # --------------------------------------------------------------------------- |
| 41 | |
| 42 | def _escalation_to_dict(rec: EscalationRecord) -> _EscalationDict: |
| 43 | return _EscalationDict( |
| 44 | escalation_id=rec.escalation_id, |
| 45 | pattern_id=rec.pattern_id, |
| 46 | reason=rec.reason, |
| 47 | escalated_at=rec.escalated_at.isoformat(), |
| 48 | escalated_by=rec.escalated_by.to_dict(), |
| 49 | resolved_at=rec.resolved_at.isoformat() if rec.resolved_at is not None else None, |
| 50 | resolved_by=rec.resolved_by.to_dict() if rec.resolved_by is not None else None, |
| 51 | resolution_id=rec.resolution_id, |
| 52 | status=rec.status, |
| 53 | ) |
| 54 | |
| 55 | def _dict_to_escalation(data: Mapping[str, JsonValue]) -> EscalationRecord: |
| 56 | escalated_by_raw = data["escalated_by"] |
| 57 | escalated_by = AgentProvenance( |
| 58 | type=escalated_by_raw.get("type", "human"), |
| 59 | agent_id=escalated_by_raw.get("agent_id"), |
| 60 | model_id=escalated_by_raw.get("model_id"), |
| 61 | ) |
| 62 | |
| 63 | resolved_at = ( |
| 64 | datetime.datetime.fromisoformat(data["resolved_at"]) |
| 65 | if data.get("resolved_at") is not None |
| 66 | else None |
| 67 | ) |
| 68 | resolved_by: AgentProvenance | None = None |
| 69 | if data.get("resolved_by") is not None: |
| 70 | rb = data["resolved_by"] |
| 71 | resolved_by = AgentProvenance( |
| 72 | type=rb.get("type", "human"), |
| 73 | agent_id=rb.get("agent_id"), |
| 74 | model_id=rb.get("model_id"), |
| 75 | ) |
| 76 | |
| 77 | return EscalationRecord( |
| 78 | escalation_id=data["escalation_id"], |
| 79 | pattern_id=data["pattern_id"], |
| 80 | reason=data["reason"], |
| 81 | escalated_at=datetime.datetime.fromisoformat(data["escalated_at"]), |
| 82 | escalated_by=escalated_by, |
| 83 | resolved_at=resolved_at, |
| 84 | resolved_by=resolved_by, |
| 85 | resolution_id=data.get("resolution_id"), |
| 86 | status=data.get("status", EscalationStatus.OPEN), |
| 87 | ) |
| 88 | |
| 89 | # --------------------------------------------------------------------------- |
| 90 | # Escalation CRUD |
| 91 | # --------------------------------------------------------------------------- |
| 92 | |
| 93 | def record_escalation(root: pathlib.Path, record: EscalationRecord) -> bool: |
| 94 | """Persist an :class:`EscalationRecord` to the harmony store. |
| 95 | |
| 96 | The file is written atomically with ``os.replace``. If the escalation |
| 97 | already exists (same ``escalation_id``), the call is a no-op and returns |
| 98 | ``False``. Returns ``True`` on first write. |
| 99 | |
| 100 | Args: |
| 101 | root: Repository root. |
| 102 | record: The escalation record to persist. |
| 103 | |
| 104 | Returns: |
| 105 | ``True`` if the record was newly written; ``False`` if it already |
| 106 | existed. |
| 107 | """ |
| 108 | esc_dir = escalations_dir(root) |
| 109 | dest = escalation_path(root, record.escalation_id) |
| 110 | dest.parent.mkdir(parents=True, exist_ok=True) |
| 111 | |
| 112 | if dest.exists() and not dest.is_symlink(): |
| 113 | return False |
| 114 | |
| 115 | payload = json.dumps(_escalation_to_dict(record), indent=2).encode() |
| 116 | fd, tmp = tempfile.mkstemp(dir=esc_dir, suffix=".tmp") |
| 117 | try: |
| 118 | os.write(fd, payload) |
| 119 | os.fsync(fd) |
| 120 | finally: |
| 121 | os.close(fd) |
| 122 | os.replace(tmp, dest) |
| 123 | logger.debug( |
| 124 | "harmony: recorded escalation %s for pattern %s", |
| 125 | short_id(record.escalation_id), |
| 126 | short_id(record.pattern_id), |
| 127 | ) |
| 128 | return True |
| 129 | |
| 130 | def load_escalation(root: pathlib.Path, escalation_id: str) -> EscalationRecord | None: |
| 131 | """Load an :class:`EscalationRecord` by ID. |
| 132 | |
| 133 | Args: |
| 134 | root: Repository root. |
| 135 | escalation_id: ``sha256:`` content-addressed escalation ID. |
| 136 | |
| 137 | Returns: |
| 138 | The :class:`EscalationRecord`, or ``None`` if not found. |
| 139 | |
| 140 | Raises: |
| 141 | ValueError: If ``escalation_id`` is not a valid content-addressed ID. |
| 142 | """ |
| 143 | _validate_id(escalation_id, "escalation_id") |
| 144 | dest = escalation_path(root, escalation_id) |
| 145 | |
| 146 | if not dest.exists() or dest.is_symlink(): |
| 147 | return None |
| 148 | |
| 149 | raw = dest.read_bytes() |
| 150 | if len(raw) > _MAX_ESCALATION_BYTES: |
| 151 | logger.warning( |
| 152 | "harmony: escalation %s exceeds size cap — skipping", escalation_id |
| 153 | ) |
| 154 | return None |
| 155 | |
| 156 | try: |
| 157 | data = json.loads(raw) |
| 158 | return _dict_to_escalation(data) |
| 159 | except Exception as exc: |
| 160 | logger.warning( |
| 161 | "harmony: failed to parse escalation %s: %s", escalation_id, exc |
| 162 | ) |
| 163 | return None |
| 164 | |
| 165 | def list_escalations( |
| 166 | root: pathlib.Path, |
| 167 | status: str | None = None, |
| 168 | ) -> list[EscalationRecord]: |
| 169 | """Return all escalation records, optionally filtered by status. |
| 170 | |
| 171 | Results are sorted newest-first by ``escalated_at``. Symlinks and files |
| 172 | that exceed :data:`_MAX_ESCALATION_BYTES` are silently skipped. |
| 173 | |
| 174 | Args: |
| 175 | root: Repository root. |
| 176 | status: If given, only records with this :class:`EscalationStatus` |
| 177 | value are returned. ``None`` returns all records. |
| 178 | |
| 179 | Returns: |
| 180 | List of :class:`EscalationRecord` instances, newest-first. |
| 181 | """ |
| 182 | esc_dir = escalations_dir(root) |
| 183 | if not esc_dir.exists(): |
| 184 | return [] |
| 185 | |
| 186 | records: list[EscalationRecord] = [] |
| 187 | for algo_dir in esc_dir.iterdir(): |
| 188 | if algo_dir.is_symlink() or not algo_dir.is_dir(): |
| 189 | continue |
| 190 | for entry in algo_dir.iterdir(): |
| 191 | if entry.is_symlink() or not entry.is_file(): |
| 192 | continue |
| 193 | if not entry.name.endswith(".json"): |
| 194 | continue |
| 195 | |
| 196 | raw = entry.read_bytes() |
| 197 | if len(raw) > _MAX_ESCALATION_BYTES: |
| 198 | logger.warning( |
| 199 | "harmony: escalation file %s exceeds size cap — skipping", |
| 200 | entry.name, |
| 201 | ) |
| 202 | continue |
| 203 | |
| 204 | try: |
| 205 | data = json.loads(raw) |
| 206 | rec = _dict_to_escalation(data) |
| 207 | except Exception as exc: |
| 208 | logger.warning( |
| 209 | "harmony: failed to parse escalation %s: %s", entry.name, exc |
| 210 | ) |
| 211 | continue |
| 212 | |
| 213 | if status is not None and rec.status != status: |
| 214 | continue |
| 215 | records.append(rec) |
| 216 | |
| 217 | records.sort(key=lambda r: r.escalated_at, reverse=True) |
| 218 | return records |
| 219 | |
| 220 | def resolve_escalation( |
| 221 | root: pathlib.Path, |
| 222 | escalation_id: str, |
| 223 | resolution_id: str, |
| 224 | resolved_by: AgentProvenance, |
| 225 | resolved_at: datetime.datetime, |
| 226 | ) -> bool: |
| 227 | """Transition an :class:`EscalationRecord` from OPEN to RESOLVED. |
| 228 | |
| 229 | Reads the existing record, updates status + resolution fields, and writes |
| 230 | atomically. If the escalation does not exist, returns ``False``. |
| 231 | |
| 232 | Args: |
| 233 | root: Repository root. |
| 234 | escalation_id: ``sha256:`` content-addressed ID of the escalation. |
| 235 | resolution_id: ``sha256:`` content-addressed ID of the resolution. |
| 236 | resolved_by: :class:`AgentProvenance` of who resolved it. |
| 237 | resolved_at: UTC-aware timestamp of the resolution. |
| 238 | |
| 239 | Returns: |
| 240 | ``True`` if the record was found and updated; ``False`` if not found. |
| 241 | |
| 242 | Raises: |
| 243 | ValueError: If ``escalation_id`` is not a valid content-addressed ID. |
| 244 | """ |
| 245 | _validate_id(escalation_id, "escalation_id") |
| 246 | existing = load_escalation(root, escalation_id) |
| 247 | if existing is None: |
| 248 | return False |
| 249 | |
| 250 | updated = dc_replace( |
| 251 | existing, |
| 252 | status=EscalationStatus.RESOLVED, |
| 253 | resolution_id=resolution_id, |
| 254 | resolved_by=resolved_by, |
| 255 | resolved_at=resolved_at, |
| 256 | ) |
| 257 | |
| 258 | dest = escalation_path(root, escalation_id) |
| 259 | payload = json.dumps(_escalation_to_dict(updated), indent=2).encode() |
| 260 | fd, tmp = tempfile.mkstemp(dir=dest.parent, suffix=".tmp") |
| 261 | try: |
| 262 | os.write(fd, payload) |
| 263 | os.fsync(fd) |
| 264 | finally: |
| 265 | os.close(fd) |
| 266 | os.replace(tmp, dest) |
| 267 | logger.debug("harmony: resolved escalation %s", short_id(escalation_id)) |
| 268 | return True |
File History
3 commits
sha256:e8214e0062ef8ef0af999937df2731655b6082781fc22bc563f58db2f42b1de9
Merge 'bump/muse-v0.2.1' into 'dev' — proposal: chore: bump…
Human
10 days ago
sha256:8de4334a98c945aace420969d389ad678aa926d4ab4e886b2ac4c4241cb3bf2b
revert: keep pyproject.toml in canonical PEP 440 form
Sonnet 4.6
patch
68 days ago
sha256:a317886dc0496c4af7b285b3e41c86c4c34ea2e79afc63b8829aadb1ada7903f
chore: bump version to 0.2.0rc15 to match musehub#113 fix release
Sonnet 4.6
patch
68 days ago