| 1 | """Pure state transitions for human review of selected shot memories.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from copy import deepcopy |
| 6 | from typing import Any |
| 7 | |
| 8 | |
| 9 | class MemoryReviewConflict(ValueError): |
| 10 | """The browser acted on a stale or incompatible review state.""" |
| 11 | |
| 12 | |
| 13 | def _assert_current( |
| 14 | review: dict[str, Any], |
| 15 | *, |
| 16 | review_id: str, |
| 17 | attempt: int, |
| 18 | ) -> None: |
| 19 | if ( |
| 20 | str(review.get("review_id") or "") != review_id |
| 21 | or int(review.get("attempt") or 0) != int(attempt) |
| 22 | ): |
| 23 | raise MemoryReviewConflict("stale memory review attempt") |
| 24 | |
| 25 | |
| 26 | def approve_memory_review( |
| 27 | review: dict[str, Any], |
| 28 | *, |
| 29 | review_id: str, |
| 30 | attempt: int, |
| 31 | updated_at: str, |
| 32 | retained_memory_ids: list[str] | None = None, |
| 33 | ) -> bool: |
| 34 | """Approve the exact proposal, returning whether state changed.""" |
| 35 | _assert_current(review, review_id=review_id, attempt=attempt) |
| 36 | status = str(review.get("status") or "") |
| 37 | if status == "approved": |
| 38 | return False |
| 39 | if status != "awaiting_review": |
| 40 | raise MemoryReviewConflict(f"memory review cannot be approved from {status}") |
| 41 | available_ids = { |
| 42 | str(item.get("memory_id") or "") |
| 43 | for item in review.get("selections", []) |
| 44 | if isinstance(item, dict) and item.get("memory_id") |
| 45 | } |
| 46 | if retained_memory_ids is None: |
| 47 | source_ids = review.get("retained_memory_ids") |
| 48 | retained_memory_ids = ( |
| 49 | list(source_ids) |
| 50 | if isinstance(source_ids, list) |
| 51 | else [ |
| 52 | str(item.get("memory_id") or "") |
| 53 | for item in review.get("selections", []) |
| 54 | if isinstance(item, dict) and item.get("memory_id") |
| 55 | ] |
| 56 | ) |
| 57 | retained = list(dict.fromkeys(str(value) for value in retained_memory_ids if value)) |
| 58 | unknown = sorted(set(retained) - available_ids) |
| 59 | if unknown: |
| 60 | raise MemoryReviewConflict( |
| 61 | "retained memory is not in this review: " + ", ".join(unknown) |
| 62 | ) |
| 63 | review["retained_memory_ids"] = retained |
| 64 | review["status"] = "approved" |
| 65 | review["error"] = None |
| 66 | review["updated_at"] = updated_at |
| 67 | return True |
| 68 | |
| 69 | |
| 70 | def reselect_memory_review( |
| 71 | review: dict[str, Any], |
| 72 | *, |
| 73 | review_id: str, |
| 74 | attempt: int, |
| 75 | updated_at: str, |
| 76 | memory_id: str | None = None, |
| 77 | ) -> bool: |
| 78 | """Reject one proposal item, or every item for legacy callers.""" |
| 79 | _assert_current(review, review_id=review_id, attempt=attempt) |
| 80 | status = str(review.get("status") or "") |
| 81 | if status == "reselecting": |
| 82 | return False |
| 83 | if status not in {"awaiting_review", "error"}: |
| 84 | raise MemoryReviewConflict(f"memory review cannot be reselected from {status}") |
| 85 | |
| 86 | selections = [ |
| 87 | item for item in review.get("selections", []) if isinstance(item, dict) |
| 88 | ] |
| 89 | if memory_id: |
| 90 | selections = [ |
| 91 | item for item in selections |
| 92 | if str(item.get("memory_id") or "") == memory_id |
| 93 | ] |
| 94 | if not selections: |
| 95 | raise MemoryReviewConflict(f"memory {memory_id} is not in this review") |
| 96 | rejected = { |
| 97 | int(value) |
| 98 | for value in review.get("rejected_candidate_indices", []) |
| 99 | if isinstance(value, int) or (isinstance(value, str) and value.isdigit()) |
| 100 | } |
| 101 | for selection in selections: |
| 102 | try: |
| 103 | rejected.add(int(selection["candidate_index"])) |
| 104 | except (KeyError, TypeError, ValueError): |
| 105 | continue |
| 106 | |
| 107 | history = review.setdefault("history", []) |
| 108 | if not isinstance(history, list): |
| 109 | history = [] |
| 110 | review["history"] = history |
| 111 | history.append( |
| 112 | { |
| 113 | "attempt": int(attempt), |
| 114 | "selections": deepcopy(selections), |
| 115 | "rejected_at": updated_at, |
| 116 | } |
| 117 | ) |
| 118 | review["rejected_candidate_indices"] = sorted(rejected) |
| 119 | review["reselect_memory_id"] = memory_id |
| 120 | review["status"] = "reselecting" |
| 121 | review["error"] = None |
| 122 | review["updated_at"] = updated_at |
| 123 | return True |
| 124 |