Find entry containing old_text substring, replace it with new_content.
(self, target: str, old_text: str, new_content: str)
| 386 | return self._success_response(target, "Entry added.") |
| 387 | |
| 388 | def replace(self, target: str, old_text: str, new_content: str) -> Dict[str, Any]: |
| 389 | """Find entry containing old_text substring, replace it with new_content.""" |
| 390 | old_text = old_text.strip() |
| 391 | new_content = new_content.strip() |
| 392 | if not old_text: |
| 393 | return {"success": False, "error": "old_text cannot be empty."} |
| 394 | if not new_content: |
| 395 | return {"success": False, "error": "new_content cannot be empty. Use 'remove' to delete entries."} |
| 396 | |
| 397 | # Scan replacement content for injection/exfiltration |
| 398 | scan_error = _scan_memory_content(new_content) |
| 399 | if scan_error: |
| 400 | return {"success": False, "error": scan_error} |
| 401 | |
| 402 | with self._file_lock(self._path_for(target)): |
| 403 | bak = self._reload_target(target) |
| 404 | if bak: |
| 405 | return _drift_error(self._path_for(target), bak) |
| 406 | |
| 407 | entries = self._entries_for(target) |
| 408 | matches = [(i, e) for i, e in enumerate(entries) if old_text in e] |
| 409 | |
| 410 | if not matches: |
| 411 | return self._consolidation_failure({ |
| 412 | "success": False, |
| 413 | "error": f"No entry matched '{old_text}'. Check current_entries below and retry with the exact text of the entry you want to replace.", |
| 414 | "current_entries": entries, |
| 415 | }) |
| 416 | |
| 417 | if len(matches) > 1: |
| 418 | # If all matches are identical (exact duplicates), operate on the first one |
| 419 | unique_texts = {e for _, e in matches} |
| 420 | if len(unique_texts) > 1: |
| 421 | previews = self._previews([e for _, e in matches]) |
| 422 | return { |
| 423 | "success": False, |
| 424 | "error": f"Multiple entries matched '{old_text}'. Be more specific.", |
| 425 | "matches": previews, |
| 426 | } |
| 427 | # All identical -- safe to replace just the first |
| 428 | |
| 429 | idx = matches[0][0] |
| 430 | limit = self._char_limit(target) |
| 431 | |
| 432 | # Check that replacement doesn't blow the budget |
| 433 | test_entries = entries.copy() |
| 434 | test_entries[idx] = new_content |
| 435 | new_total = len(ENTRY_DELIMITER.join(test_entries)) |
| 436 | |
| 437 | if new_total > limit: |
| 438 | current = self._char_count(target) |
| 439 | return self._consolidation_failure({ |
| 440 | "success": False, |
| 441 | "error": ( |
| 442 | f"Replacement would put memory at {new_total:,}/{limit:,} chars. " |
| 443 | f"Shorten the new content, or 'remove' other stale or less important " |
| 444 | f"entries to make room (see current_entries below), then retry — all " |
| 445 | f"in this turn." |