| 1 | """Track file-read state for read-before-edit warnings and read deduplication.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import hashlib |
| 6 | import os |
| 7 | from dataclasses import dataclass |
| 8 | from pathlib import Path |
| 9 | |
| 10 | |
| 11 | @dataclass(slots=True) |
| 12 | class ReadState: |
| 13 | mtime: float |
| 14 | offset: int |
| 15 | limit: int | None |
| 16 | content_hash: str | None |
| 17 | can_dedup: bool |
| 18 | |
| 19 | |
| 20 | _state: dict[str, ReadState] = {} |
| 21 | |
| 22 | |
| 23 | def _hash_file(p: str) -> str | None: |
| 24 | try: |
| 25 | return hashlib.sha256(Path(p).read_bytes()).hexdigest() |
| 26 | except OSError: |
| 27 | return None |
| 28 | |
| 29 | |
| 30 | def record_read(path: str | Path, offset: int = 1, limit: int | None = None) -> None: |
| 31 | """Record that a file was read (called after successful read).""" |
| 32 | p = str(Path(path).resolve()) |
| 33 | try: |
| 34 | mtime = os.path.getmtime(p) |
| 35 | except OSError: |
| 36 | return |
| 37 | _state[p] = ReadState( |
| 38 | mtime=mtime, |
| 39 | offset=offset, |
| 40 | limit=limit, |
| 41 | content_hash=_hash_file(p), |
| 42 | can_dedup=True, |
| 43 | ) |
| 44 | |
| 45 | |
| 46 | def record_write(path: str | Path) -> None: |
| 47 | """Record that a file was written (updates mtime in state).""" |
| 48 | p = str(Path(path).resolve()) |
| 49 | try: |
| 50 | mtime = os.path.getmtime(p) |
| 51 | except OSError: |
| 52 | _state.pop(p, None) |
| 53 | return |
| 54 | _state[p] = ReadState( |
| 55 | mtime=mtime, |
| 56 | offset=1, |
| 57 | limit=None, |
| 58 | content_hash=_hash_file(p), |
| 59 | can_dedup=False, |
| 60 | ) |
| 61 | |
| 62 | |
| 63 | def check_read(path: str | Path) -> str | None: |
| 64 | """Check if a file has been read and is fresh. |
| 65 | |
| 66 | Returns None if OK, or a warning string. |
| 67 | When mtime changed but file content is identical (e.g. touch, editor save), |
| 68 | the check passes to avoid false-positive staleness warnings. |
| 69 | """ |
| 70 | p = str(Path(path).resolve()) |
| 71 | entry = _state.get(p) |
| 72 | if entry is None: |
| 73 | return "Warning: file has not been read yet. Read it first to verify content before editing." |
| 74 | try: |
| 75 | current_mtime = os.path.getmtime(p) |
| 76 | except OSError: |
| 77 | return None |
| 78 | if current_mtime != entry.mtime: |
| 79 | if entry.content_hash and _hash_file(p) == entry.content_hash: |
| 80 | entry.mtime = current_mtime |
| 81 | return None |
| 82 | return "Warning: file has been modified since last read. Re-read to verify content before editing." |
| 83 | # mtime unchanged - still check content hash to detect quick modifications |
| 84 | if entry.content_hash and _hash_file(p) != entry.content_hash: |
| 85 | return "Warning: file has been modified since last read. Re-read to verify content before editing." |
| 86 | return None |
| 87 | |
| 88 | |
| 89 | def is_unchanged(path: str | Path, offset: int = 1, limit: int | None = None) -> bool: |
| 90 | """Return True if file was previously read with same params and content is unchanged.""" |
| 91 | p = str(Path(path).resolve()) |
| 92 | entry = _state.get(p) |
| 93 | if entry is None: |
| 94 | return False |
| 95 | if not entry.can_dedup: |
| 96 | return False |
| 97 | if entry.offset != offset or entry.limit != limit: |
| 98 | return False |
| 99 | try: |
| 100 | current_mtime = os.path.getmtime(p) |
| 101 | except OSError: |
| 102 | return False |
| 103 | if current_mtime != entry.mtime: |
| 104 | # mtime changed - check if content also changed |
| 105 | current_hash = _hash_file(p) |
| 106 | if current_hash != entry.content_hash: |
| 107 | # Content actually changed - don't dedup |
| 108 | entry.can_dedup = False |
| 109 | return False |
| 110 | # Content identical despite mtime change (e.g. touch) - mark as not dedupable to force full read next time |
| 111 | entry.can_dedup = False |
| 112 | return True |
| 113 | # mtime unchanged - content must be identical |
| 114 | return True |
| 115 | |
| 116 | |
| 117 | def clear() -> None: |
| 118 | """Clear all tracked state (useful for testing).""" |
| 119 | _state.clear() |
| 120 |