Fix mypy type errors in transaction_manager.py and transactions.py

Author: codegen-sh[bot]

codegen-on-oss/codegen_on_oss/analyzers/transaction_manager.py

@@ -61,9 +61,9 @@ def __init__(self) -> None:
         self.pending_undos: Set[Callable[[], None]] = set()
         self._commiting: bool = False
         self.max_transactions: Optional[int] = None  # None = no limit
-        self.stopwatch_start = None
+        self.stopwatch_start: Optional[float] = None
         self.stopwatch_max_seconds: Optional[int] = None  # None = no limit
-        self.session = {}  # Session data for tracking state
+        self.session: Dict[str, Any] = {}  # Session data for tracking state
 
     def sort_transactions(self) -> None:
         """Sort transactions by priority and position."""
@@ -127,7 +127,7 @@ def reset_stopwatch(self, max_seconds: Optional[int] = None) -> None:
 
     def is_time_exceeded(self) -> bool:
         """Check if the stopwatch time limit has been exceeded."""
-        if self.stopwatch_max_seconds is None:
+        if self.stopwatch_max_seconds is None or self.stopwatch_start is None:
             return False
         else:
             num_seconds = time.time() - self.stopwatch_start
@@ -384,7 +384,7 @@ def get_transactions_at_range(self, file_path: Path, start_byte: int, end_byte:
         Returns:
             List of matching transactions
         """
-        matching_transactions = []
+        matching_transactions: List[Transaction] = []
         if file_path not in self.queued_transactions:
             return matching_transactions
 
@@ -435,14 +435,24 @@ def _get_conflicts(self, transaction: Transaction) -> List[Transaction]:
         Returns:
             List of conflicting transactions
         """
-        overlapping_transactions = []
+        overlapping_transactions: List[Transaction] = []
         if transaction.file_path not in self.queued_transactions:
             return overlapping_transactions
-            
-        queued_transactions = list(self.queued_transactions[transaction.file_path])
-        for t in queued_transactions:
-            if transaction.start_byte < t.end_byte and transaction.end_byte > t.start_byte:
+
+        for t in self.queued_transactions[transaction.file_path]:
+            # Skip if it's the same transaction
+            if t == transaction:
+                continue
+
+            # Check if the transactions overlap
+            if (
+                (t.start_byte <= transaction.start_byte < t.end_byte)
+                or (t.start_byte < transaction.end_byte <= t.end_byte)
+                or (transaction.start_byte <= t.start_byte < transaction.end_byte)
+                or (transaction.start_byte < t.end_byte <= transaction.end_byte)
+            ):
                 overlapping_transactions.append(t)
+
         return overlapping_transactions
 
     def _get_overlapping_conflicts(self, transaction: Transaction) -> Optional[Transaction]:
@@ -456,9 +466,8 @@ def _get_overlapping_conflicts(self, transaction: Transaction) -> Optional[Trans
         """
         if transaction.file_path not in self.queued_transactions:
             return None
-            
+
         for t in self.queued_transactions[transaction.file_path]:
             if transaction.start_byte >= t.start_byte and transaction.end_byte <= t.end_byte:
                 return t
         return None
-

codegen-on-oss/codegen_on_oss/analyzers/transactions.py

@@ -210,6 +210,8 @@ def __init__(
 
     def _generate_new_content_bytes(self) -> bytes:
         """Generate the new content bytes after insertion."""
+        if self.new_content is None:
+            raise ValueError("Cannot generate content bytes: new_content is None")
         new_bytes = bytes(self.new_content, encoding="utf-8")
         content_bytes = self.file.content_bytes
         head = content_bytes[: self.insert_byte]
@@ -230,7 +232,8 @@ def get_diff(self) -> DiffLite:
     def diff_str(self) -> str:
         """Human-readable string representation of the change."""
         diff = "".join(unified_diff(self.file.content.splitlines(True), self._generate_new_content_bytes().decode("utf-8").splitlines(True)))
-        return f"Insert {len(self.new_content)} bytes at bytes ({self.start_byte}, {self.end_byte})\n{diff}"
+        content_length = len(self.new_content) if self.new_content is not None else 0
+        return f"Insert {content_length} bytes at bytes ({self.start_byte}, {self.end_byte})\n{diff}"
 
 class EditTransaction(Transaction):
     """Transaction to edit content in a file."""
@@ -364,4 +367,3 @@ def get_diff(self) -> DiffLite:
     def diff_str(self) -> str:
         """Human-readable string representation of the change."""
         return f"Remove file at {self.file_path}"
-