+54

think/activities.py

reviewed

··· 1153 1153 return updated 1154 1154 1155 1155 1156 1156 + def append_ledger_close_edit( 1157 1157 + facet: str, 1158 1158 + day: str, 1159 1159 + record_id: str, 1160 1160 + *, 1161 1161 + item_id: str, 1162 1162 + note: str, 1163 1163 + as_state: str, 1164 1164 + ) -> dict[str, Any] | None: 1165 1165 + """Append one ledger-close audit edit to an activity record.""" 1166 1166 + updated_record: dict[str, Any] | None = None 1167 1167 + 1168 1168 + def modify_fn(records: list[dict[str, Any]]) -> list[dict[str, Any]]: 1169 1169 + nonlocal updated_record 1170 1170 + new_records: list[dict[str, Any]] = [] 1171 1171 + for record in records: 1172 1172 + if record.get("id") != record_id: 1173 1173 + new_records.append(record) 1174 1174 + continue 1175 1175 + 1176 1176 + normalized = _normalize_activity_record(record) 1177 1177 + edits = normalized.get("edits", []) 1178 1178 + already_closed = any( 1179 1179 + edit.get("fields") == ["ledger_close"] 1180 1180 + and isinstance(edit.get("ledger_close"), dict) 1181 1181 + and edit["ledger_close"].get("item_id") == item_id 1182 1182 + and edit["ledger_close"].get("as_state") == as_state 1183 1183 + for edit in edits 1184 1184 + if isinstance(edit, dict) 1185 1185 + ) 1186 1186 + if already_closed: 1187 1187 + updated_record = normalized 1188 1188 + new_records.append(normalized) 1189 1189 + continue 1190 1190 + 1191 1191 + normalized = append_edit( 1192 1192 + normalized, 1193 1193 + actor="cli:ledger_close", 1194 1194 + fields=["ledger_close"], 1195 1195 + note=note, 1196 1196 + payload={"ledger_close": {"item_id": item_id, "as_state": as_state}}, 1197 1197 + ) 1198 1198 + updated_record = normalized 1199 1199 + new_records.append(normalized) 1200 1200 + return new_records 1201 1201 + 1202 1202 + try: 1203 1203 + locked_modify(_get_records_path(facet, day), modify_fn) 1204 1204 + except FileNotFoundError: 1205 1205 + return None 1206 1206 + 1207 1207 + return updated_record 1208 1208 + 1209 1209 + 1156 1210 def _set_activity_hidden_state( 1157 1211 facet: str, 1158 1212 day: str,

+593

think/surfaces/ledger.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc 3 3 + 4 4 + """Ledger surface for commitments, closures, and decisions. 5 5 + 6 6 + Dropped/deferred talent resolutions: any matched closure -> state="closed" 7 7 + regardless of its `resolution` field. CLI `--as dropped` is the only path to 8 8 + state="dropped". 9 9 + """ 10 10 + 11 11 + from __future__ import annotations 12 12 + 13 13 + import builtins 14 14 + import hashlib 15 15 + import re 16 16 + from datetime import UTC, datetime 17 17 + from pathlib import Path 18 18 + from typing import Any, Iterable, Iterator 19 19 + 20 20 + from think.activities import append_ledger_close_edit, load_activity_records 21 21 + from think.entities.matching import find_matching_entity 22 22 + from think.facets import get_enabled_facets, get_facets 23 23 + from think.surfaces.types import ActivitySourceRef, Decision, LedgerItem 24 24 + from think.utils import get_journal 25 25 + 26 26 + ACTION_MATCH_THRESHOLD = 78 27 27 + # 78 < entity-name default 90: action phrases are longer/more variable (tense, articles) and need looser matching without opening to semantically unrelated strings. 28 28 + _DAY_RE = re.compile(r"^\d{8}$") 29 29 + _FIELD_ORDER = {"commitments": 0, "closures": 1, "decisions": 2, "edits": 3} 30 30 + _ONE_DAY_MS = 86_400_000 31 31 + 32 32 + 33 33 + def _normalize_action(s: str) -> str: 34 34 + # Lowercase, trim, and collapse whitespace so dedup/fuzzy matching stay stable without stripping semantic tokens. 35 35 + return " ".join(str(s).strip().lower().split()) 36 36 + 37 37 + 38 38 + def _normalize_text(s: str | None) -> str: 39 39 + return " ".join(str(s or "").strip().lower().split()) 40 40 + 41 41 + 42 42 + def _dedup_key( 43 43 + owner_entity_id: str | None, 44 44 + action_normalized: str, 45 45 + counterparty_entity_id: str | None, 46 46 + ) -> str: 47 47 + digest = hashlib.sha256( 48 48 + f"{owner_entity_id or ''}|{action_normalized}|{counterparty_entity_id or ''}".encode( 49 49 + "utf-8" 50 50 + ) 51 51 + ) 52 52 + return digest.hexdigest()[:16] 53 53 + 54 54 + 55 55 + def _decision_key(owner_entity_id: str | None, action_normalized: str, day: str) -> str: 56 56 + digest = hashlib.sha256( 57 57 + f"{owner_entity_id or ''}|{action_normalized}|{day}".encode("utf-8") 58 58 + ) 59 59 + return digest.hexdigest()[:16] 60 60 + 61 61 + 62 62 + def _record_created_at(record: dict[str, Any]) -> int: 63 63 + return int(record.get("created_at", 0) or 0) 64 64 + 65 65 + 66 66 + def _activity_days(facet: str) -> builtins.list[str]: 67 67 + activities_dir = Path(get_journal()) / "facets" / facet / "activities" 68 68 + if not activities_dir.is_dir(): 69 69 + return [] 70 70 + return sorted( 71 71 + path.stem 72 72 + for path in activities_dir.glob("*.jsonl") 73 73 + if path.is_file() and _DAY_RE.fullmatch(path.stem) 74 74 + ) 75 75 + 76 76 + 77 77 + def _scan_records(facets: Iterable[str]) -> Iterator[tuple[str, str, dict[str, Any]]]: 78 78 + for facet in facets: 79 79 + for day in _activity_days(facet): 80 80 + for record in load_activity_records(facet, day): 81 81 + yield facet, day, record 82 82 + 83 83 + 84 84 + def _source_ref( 85 85 + *, facet: str, day: str, activity_id: str, field: str, created_at: int 86 86 + ) -> ActivitySourceRef: 87 87 + return ActivitySourceRef( 88 88 + facet=facet, 89 89 + day=day, 90 90 + activity_id=activity_id, 91 91 + field=field, 92 92 + created_at=created_at, 93 93 + ) 94 94 + 95 95 + 96 96 + def _source_sort_key(source: ActivitySourceRef) -> tuple[int, str, str, str, int]: 97 97 + return ( 98 98 + source.created_at, 99 99 + source.facet, 100 100 + source.day, 101 101 + source.activity_id, 102 102 + _FIELD_ORDER.get(source.field, 99), 103 103 + ) 104 104 + 105 105 + 106 106 + def _chronological_key( 107 107 + created_at: int, facet: str, day: str, activity_id: str 108 108 + ) -> tuple[int, str, str, str]: 109 109 + return created_at, facet, day, activity_id 110 110 + 111 111 + 112 112 + def _edit_timestamp_ms(edit: dict[str, Any], fallback: int) -> int: 113 113 + timestamp = edit.get("timestamp") 114 114 + if not isinstance(timestamp, str) or not timestamp: 115 115 + return fallback 116 116 + try: 117 117 + parsed = datetime.fromisoformat(timestamp.replace("Z", "+00:00")) 118 118 + except ValueError: 119 119 + return fallback 120 120 + return int(parsed.timestamp() * 1000) 121 121 + 122 122 + 123 123 + def _actions_match(commitment_action: str, candidate_action: str) -> bool: 124 124 + if not commitment_action or not candidate_action: 125 125 + return False 126 126 + # Reuse the entity matcher with throwaway action "entities" so ledger pairing stays on the same fuzzy matching surface as entity resolution. 127 127 + match = find_matching_entity( 128 128 + commitment_action, 129 129 + [{"name": candidate_action}], 130 130 + fuzzy_threshold=ACTION_MATCH_THRESHOLD, 131 131 + ) 132 132 + return match is not None 133 133 + 134 134 + 135 135 + def _entity_pair_matches( 136 136 + left_id: str | None, right_id: str | None, *, allow_both_missing: bool 137 137 + ) -> bool: 138 138 + if left_id and right_id: 139 139 + return left_id == right_id 140 140 + if left_id or right_id: 141 141 + return False 142 142 + return allow_both_missing 143 143 + 144 144 + 145 145 + def _counterparty_matches(item: dict[str, Any], closure: dict[str, Any]) -> bool: 146 146 + item_id = item["counterparty_entity_id"] 147 147 + closure_id = closure["counterparty_entity_id"] 148 148 + if item_id and closure_id: 149 149 + return item_id == closure_id 150 150 + if item_id or closure_id: 151 151 + return False 152 152 + return item["counterparty_normalized"] == closure["counterparty_normalized"] 153 153 + 154 154 + 155 155 + def _story_closure_matches(item: dict[str, Any], closure: dict[str, Any]) -> bool: 156 156 + if not _entity_pair_matches( 157 157 + item["owner_entity_id"], closure["owner_entity_id"], allow_both_missing=True 158 158 + ): 159 159 + return False 160 160 + if not _counterparty_matches(item, closure): 161 161 + return False 162 162 + return _actions_match(item["action_normalized"], closure["action_normalized"]) 163 163 + 164 164 + 165 165 + def _resolve_sort(state: str, sort: str | None) -> str: 166 166 + valid = {"age_days_desc", "opened_at_desc", "closed_at_desc"} 167 167 + if sort is not None and sort not in valid: 168 168 + raise ValueError(f"unknown sort: {sort}") 169 169 + if sort is not None: 170 170 + return sort 171 171 + if state in {"closed", "dropped"}: 172 172 + return "closed_at_desc" 173 173 + return "age_days_desc" 174 174 + 175 175 + 176 176 + def _validate_state(state: str) -> str: 177 177 + if state not in {"open", "closed", "dropped", "all"}: 178 178 + raise ValueError(f"unknown state: {state}") 179 179 + return state 180 180 + 181 181 + 182 182 + def _party_matches(query: str, name: str | None, entity_id: str | None) -> bool: 183 183 + normalized_query = _normalize_text(query) 184 184 + if not normalized_query: 185 185 + return True 186 186 + candidates = [name or "", entity_id or "", (entity_id or "").replace("_", " ")] 187 187 + return any( 188 188 + normalized_query in _normalize_text(candidate) for candidate in candidates 189 189 + ) 190 190 + 191 191 + 192 192 + def _parse_day_ms(day: str, *, field_name: str) -> int: 193 193 + try: 194 194 + parsed = datetime.strptime(day, "%Y%m%d").replace(tzinfo=UTC) 195 195 + except ValueError as exc: 196 196 + raise ValueError(f"{field_name} must match YYYYMMDD") from exc 197 197 + return int(parsed.timestamp() * 1000) 198 198 + 199 199 + 200 200 + def _list_all_facets() -> builtins.list[str]: 201 201 + return builtins.list(get_facets().keys()) 202 202 + 203 203 + 204 204 + def _build_ledger_items( 205 205 + records: Iterable[tuple[str, str, dict[str, Any]]], 206 206 + ) -> builtins.list[LedgerItem]: 207 207 + now_ms = int(datetime.now(UTC).timestamp() * 1000) 208 208 + commitments: dict[str, dict[str, Any]] = {} 209 209 + story_closures: builtins.list[dict[str, Any]] = [] 210 210 + manual_closes: dict[str, builtins.list[dict[str, Any]]] = {} 211 211 + 212 212 + for facet, day, record in records: 213 213 + record_id = str(record.get("id") or "") 214 214 + if not record_id: 215 215 + continue 216 216 + created_at = _record_created_at(record) 217 217 + 218 218 + for raw_commitment in record.get("commitments", []): 219 219 + if not isinstance(raw_commitment, dict): 220 220 + continue 221 221 + owner = str(raw_commitment.get("owner") or "").strip() 222 222 + action = str(raw_commitment.get("action") or "").strip() 223 223 + if not owner or not action: 224 224 + continue 225 225 + 226 226 + owner_entity_id = raw_commitment.get("owner_entity_id") 227 227 + if not isinstance(owner_entity_id, str): 228 228 + owner_entity_id = None 229 229 + counterparty = str(raw_commitment.get("counterparty") or "").strip() or None 230 230 + counterparty_entity_id = raw_commitment.get("counterparty_entity_id") 231 231 + if not isinstance(counterparty_entity_id, str): 232 232 + counterparty_entity_id = None 233 233 + action_normalized = _normalize_action(action) 234 234 + item_id = _dedup_key( 235 235 + owner_entity_id, action_normalized, counterparty_entity_id 236 236 + ) 237 237 + source = _source_ref( 238 238 + facet=facet, 239 239 + day=day, 240 240 + activity_id=record_id, 241 241 + field="commitments", 242 242 + created_at=created_at, 243 243 + ) 244 244 + opening_key = _chronological_key(created_at, facet, day, record_id) 245 245 + entry = commitments.get(item_id) 246 246 + if entry is None: 247 247 + commitments[item_id] = { 248 248 + "id": item_id, 249 249 + "owner": owner, 250 250 + "owner_entity_id": owner_entity_id, 251 251 + "counterparty": counterparty, 252 252 + "counterparty_entity_id": counterparty_entity_id, 253 253 + "counterparty_normalized": _normalize_text(counterparty), 254 254 + "action": action, 255 255 + "action_normalized": action_normalized, 256 256 + "when": str(raw_commitment.get("when") or "").strip() or None, 257 257 + "context": str(raw_commitment.get("context") or ""), 258 258 + "opened_at": created_at, 259 259 + "opening_key": opening_key, 260 260 + "sources": [source], 261 261 + } 262 262 + continue 263 263 + 264 264 + entry["sources"].append(source) 265 265 + if opening_key < entry["opening_key"]: 266 266 + entry.update( 267 267 + { 268 268 + "owner": owner, 269 269 + "owner_entity_id": owner_entity_id, 270 270 + "counterparty": counterparty, 271 271 + "counterparty_entity_id": counterparty_entity_id, 272 272 + "counterparty_normalized": _normalize_text(counterparty), 273 273 + "action": action, 274 274 + "action_normalized": action_normalized, 275 275 + "when": str(raw_commitment.get("when") or "").strip() or None, 276 276 + "context": str(raw_commitment.get("context") or ""), 277 277 + "opened_at": created_at, 278 278 + "opening_key": opening_key, 279 279 + } 280 280 + ) 281 281 + 282 282 + for raw_closure in record.get("closures", []): 283 283 + if not isinstance(raw_closure, dict): 284 284 + continue 285 285 + action = str(raw_closure.get("action") or "").strip() 286 286 + if not action: 287 287 + continue 288 288 + owner_entity_id = raw_closure.get("owner_entity_id") 289 289 + if not isinstance(owner_entity_id, str): 290 290 + owner_entity_id = None 291 291 + counterparty_entity_id = raw_closure.get("counterparty_entity_id") 292 292 + if not isinstance(counterparty_entity_id, str): 293 293 + counterparty_entity_id = None 294 294 + story_closures.append( 295 295 + { 296 296 + "owner_entity_id": owner_entity_id, 297 297 + "counterparty_entity_id": counterparty_entity_id, 298 298 + "counterparty_normalized": _normalize_text( 299 299 + str(raw_closure.get("counterparty") or "").strip() or None 300 300 + ), 301 301 + "action_normalized": _normalize_action(action), 302 302 + "closed_at": created_at, 303 303 + "state": "closed", 304 304 + "sort_key": _chronological_key(created_at, facet, day, record_id), 305 305 + "source": _source_ref( 306 306 + facet=facet, 307 307 + day=day, 308 308 + activity_id=record_id, 309 309 + field="closures", 310 310 + created_at=created_at, 311 311 + ), 312 312 + } 313 313 + ) 314 314 + 315 315 + for raw_edit in record.get("edits", []): 316 316 + if not isinstance(raw_edit, dict): 317 317 + continue 318 318 + if raw_edit.get("fields") != ["ledger_close"]: 319 319 + continue 320 320 + ledger_close = raw_edit.get("ledger_close") 321 321 + if not isinstance(ledger_close, dict): 322 322 + continue 323 323 + item_id = ledger_close.get("item_id") 324 324 + as_state = ledger_close.get("as_state") 325 325 + if not isinstance(item_id, str) or not item_id: 326 326 + continue 327 327 + if as_state not in {"closed", "dropped"}: 328 328 + continue 329 329 + closed_at = _edit_timestamp_ms(raw_edit, created_at) 330 330 + manual_closes.setdefault(item_id, []).append( 331 331 + { 332 332 + "closed_at": closed_at, 333 333 + "state": as_state, 334 334 + "sort_key": _chronological_key(closed_at, facet, day, record_id), 335 335 + "source": _source_ref( 336 336 + facet=facet, 337 337 + day=day, 338 338 + activity_id=record_id, 339 339 + field="edits", 340 340 + created_at=created_at, 341 341 + ), 342 342 + } 343 343 + ) 344 344 + 345 345 + story_closures.sort(key=lambda candidate: candidate["sort_key"]) 346 346 + consumed_story_closures: set[int] = set() 347 347 + items: builtins.list[LedgerItem] = [] 348 348 + 349 349 + for entry in sorted(commitments.values(), key=lambda item: item["opening_key"]): 350 350 + matched_story_sources: builtins.list[dict[str, Any]] = [] 351 351 + for index, candidate in enumerate(story_closures): 352 352 + if index in consumed_story_closures: 353 353 + continue 354 354 + if not _story_closure_matches(entry, candidate): 355 355 + continue 356 356 + matched_story_sources.append(candidate) 357 357 + consumed_story_closures.add(index) 358 358 + 359 359 + closure_sources = matched_story_sources + manual_closes.get(entry["id"], []) 360 360 + closure_sources.sort(key=lambda candidate: candidate["sort_key"]) 361 361 + first_close = closure_sources[0] if closure_sources else None 362 362 + state = first_close["state"] if first_close is not None else "open" 363 363 + closed_at = first_close["closed_at"] if first_close is not None else None 364 364 + sources = builtins.list(entry["sources"]) 365 365 + sources.extend(candidate["source"] for candidate in closure_sources) 366 366 + sources.sort(key=_source_sort_key) 367 367 + 368 368 + items.append( 369 369 + LedgerItem( 370 370 + id=entry["id"], 371 371 + state=state, 372 372 + owner=entry["owner"], 373 373 + owner_entity_id=entry["owner_entity_id"], 374 374 + counterparty=entry["counterparty"], 375 375 + counterparty_entity_id=entry["counterparty_entity_id"], 376 376 + action=entry["action"], 377 377 + summary=entry["action"], 378 378 + when=entry["when"], 379 379 + context=entry["context"], 380 380 + opened_at=entry["opened_at"], 381 381 + closed_at=closed_at, 382 382 + age_days=(now_ms - entry["opened_at"]) // _ONE_DAY_MS, 383 383 + sources=tuple(sources), 384 384 + ) 385 385 + ) 386 386 + 387 387 + return items 388 388 + 389 389 + 390 390 + def _sort_items( 391 391 + items: builtins.list[LedgerItem], sort: str 392 392 + ) -> builtins.list[LedgerItem]: 393 393 + if sort == "age_days_desc": 394 394 + return sorted( 395 395 + items, 396 396 + key=lambda item: (item.age_days, item.opened_at, item.id), 397 397 + reverse=True, 398 398 + ) 399 399 + if sort == "opened_at_desc": 400 400 + return sorted(items, key=lambda item: (item.opened_at, item.id), reverse=True) 401 401 + if sort == "closed_at_desc": 402 402 + return sorted( 403 403 + items, 404 404 + key=lambda item: ( 405 405 + item.closed_at is not None, 406 406 + item.closed_at or -1, 407 407 + item.id, 408 408 + ), 409 409 + reverse=True, 410 410 + ) 411 411 + raise ValueError(f"unknown sort: {sort}") 412 412 + 413 413 + 414 414 + def list( 415 415 + *, 416 416 + state: str = "open", 417 417 + owner: str | None = None, 418 418 + counterparty: str | None = None, 419 419 + age_days_gte: int | None = None, 420 420 + closed_since: str | None = None, 421 421 + top: int | None = None, 422 422 + sort: str | None = None, 423 423 + facets: Iterable[str] | None = None, 424 424 + ) -> builtins.list[LedgerItem]: 425 425 + state = _validate_state(state) 426 426 + resolved_sort = _resolve_sort(state, sort) 427 427 + if facets is None: 428 428 + facets = get_enabled_facets().keys() # Mirror enabled-facets convention: muted facets are opted out of downstream surfaces. 429 429 + items = _build_ledger_items(_scan_records(facets)) 430 430 + 431 431 + if state != "all": 432 432 + items = [item for item in items if item.state == state] 433 433 + if owner: 434 434 + items = [ 435 435 + item 436 436 + for item in items 437 437 + if _party_matches(owner, item.owner, item.owner_entity_id) 438 438 + ] 439 439 + if counterparty: 440 440 + items = [ 441 441 + item 442 442 + for item in items 443 443 + if _party_matches( 444 444 + counterparty, item.counterparty, item.counterparty_entity_id 445 445 + ) 446 446 + ] 447 447 + if age_days_gte is not None: 448 448 + items = [item for item in items if item.age_days >= age_days_gte] 449 449 + if closed_since is not None: 450 450 + threshold_ms = _parse_day_ms(closed_since, field_name="closed_since") 451 451 + items = [ 452 452 + item 453 453 + for item in items 454 454 + if item.closed_at is not None and item.closed_at >= threshold_ms 455 455 + ] 456 456 + 457 457 + items = _sort_items(items, resolved_sort) 458 458 + if top is not None: 459 459 + items = items[:top] 460 460 + return items 461 461 + 462 462 + 463 463 + def get(item_id: str) -> LedgerItem | None: 464 464 + for item in _build_ledger_items(_scan_records(_list_all_facets())): 465 465 + if item.id == item_id: 466 466 + return item 467 467 + return None 468 468 + 469 469 + 470 470 + def close(item_id: str, *, note: str, as_state: str = "closed") -> LedgerItem: 471 471 + if as_state not in {"closed", "dropped"}: 472 472 + raise ValueError("as_state must be 'closed' or 'dropped'") 473 473 + if not note.strip(): 474 474 + raise ValueError("note must be non-empty") 475 475 + 476 476 + item = get(item_id) 477 477 + if item is None: 478 478 + raise KeyError(item_id) 479 479 + 480 480 + commitment_sources = sorted( 481 481 + (source for source in item.sources if source.field == "commitments"), 482 482 + key=_source_sort_key, 483 483 + ) 484 484 + if not commitment_sources: 485 485 + raise KeyError(item_id) 486 486 + 487 487 + source = commitment_sources[0] 488 488 + updated = append_ledger_close_edit( 489 489 + source.facet, 490 490 + source.day, 491 491 + source.activity_id, 492 492 + item_id=item_id, 493 493 + note=note.strip(), 494 494 + as_state=as_state, 495 495 + ) 496 496 + if updated is None: 497 497 + raise KeyError(item_id) 498 498 + 499 499 + refreshed = get(item_id) 500 500 + if refreshed is None: 501 501 + raise KeyError(item_id) 502 502 + return refreshed 503 503 + 504 504 + 505 505 + def decisions( 506 506 + *, 507 507 + owner: str | None = None, 508 508 + since: str | None = None, 509 509 + involving: str | None = None, 510 510 + top: int | None = None, 511 511 + facets: Iterable[str] | None = None, 512 512 + ) -> builtins.list[Decision]: 513 513 + if facets is None: 514 514 + facets = get_enabled_facets().keys() # Mirror enabled-facets convention: muted facets are opted out of downstream surfaces. 515 515 + if since is not None: 516 516 + _parse_day_ms(since, field_name="since") 517 517 + 518 518 + deduped: dict[str, Decision] = {} 519 519 + for facet, day, record in _scan_records(facets): 520 520 + record_id = str(record.get("id") or "") 521 521 + if not record_id: 522 522 + continue 523 523 + created_at = _record_created_at(record) 524 524 + source = _source_ref( 525 525 + facet=facet, 526 526 + day=day, 527 527 + activity_id=record_id, 528 528 + field="decisions", 529 529 + created_at=created_at, 530 530 + ) 531 531 + for raw_decision in record.get("decisions", []): 532 532 + if not isinstance(raw_decision, dict): 533 533 + continue 534 534 + owner_name = str(raw_decision.get("owner") or "").strip() 535 535 + action = str(raw_decision.get("action") or "").strip() 536 536 + if not owner_name or not action: 537 537 + continue 538 538 + owner_entity_id = raw_decision.get("owner_entity_id") 539 539 + if not isinstance(owner_entity_id, str): 540 540 + owner_entity_id = None 541 541 + decision_id = _decision_key(owner_entity_id, _normalize_action(action), day) 542 542 + candidate = Decision( 543 543 + id=decision_id, 544 544 + owner=owner_name, 545 545 + owner_entity_id=owner_entity_id, 546 546 + action=action, 547 547 + context=str(raw_decision.get("context") or ""), 548 548 + day=day, 549 549 + created_at=created_at, 550 550 + source=source, 551 551 + ) 552 552 + current = deduped.get(decision_id) 553 553 + if current is None or _chronological_key( 554 554 + candidate.created_at, 555 555 + candidate.source.facet, 556 556 + candidate.day, 557 557 + candidate.source.activity_id, 558 558 + ) < _chronological_key( 559 559 + current.created_at, 560 560 + current.source.facet, 561 561 + current.day, 562 562 + current.source.activity_id, 563 563 + ): 564 564 + deduped[decision_id] = candidate 565 565 + 566 566 + results = builtins.list(deduped.values()) 567 567 + if owner: 568 568 + results = [ 569 569 + decision 570 570 + for decision in results 571 571 + if _party_matches(owner, decision.owner, decision.owner_entity_id) 572 572 + ] 573 573 + if involving: 574 574 + results = [ 575 575 + decision 576 576 + for decision in results 577 577 + if _party_matches(involving, decision.owner, decision.owner_entity_id) 578 578 + ] 579 579 + if since: 580 580 + results = [decision for decision in results if decision.day >= since] 581 581 + 582 582 + results.sort( 583 583 + key=lambda decision: ( 584 584 + decision.created_at, 585 585 + decision.source.facet, 586 586 + decision.day, 587 587 + decision.source.activity_id, 588 588 + ), 589 589 + reverse=True, 590 590 + ) 591 591 + if top is not None: 592 592 + results = results[:top] 593 593 + return results