diff --git a/database/operations.py b/database/operations.py index ce48a29..0c27b83 100644 --- a/database/operations.py +++ b/database/operations.py @@ -190,6 +190,7 @@ def search_prompts( folder: Optional[str] = None, limit: int = 100, offset: int = 0, + tag_partial: bool = False, ) -> List[Dict[str, Any]]: """ Search prompts with various filters. @@ -205,6 +206,7 @@ def search_prompts( folder: Filter by subfolder name in generated image paths limit: Maximum number of results offset: Number of results to skip + tag_partial: Use LIKE matching for tags instead of exact match Returns: List of dictionaries containing prompt data @@ -222,12 +224,20 @@ def search_prompts( if tags: for tag in tags: - query_parts.append( - "AND prompts.id IN (" - " SELECT pt.prompt_id FROM prompt_tags pt" - " JOIN tags t ON pt.tag_id = t.id WHERE t.name = ?)" - ) - params.append(tag) + if tag_partial: + query_parts.append( + "AND prompts.id IN (" + " SELECT pt.prompt_id FROM prompt_tags pt" + " JOIN tags t ON pt.tag_id = t.id WHERE t.name LIKE ?)" + ) + params.append(f"%{tag}%") + else: + query_parts.append( + "AND prompts.id IN (" + " SELECT pt.prompt_id FROM prompt_tags pt" + " JOIN tags t ON pt.tag_id = t.id WHERE t.name = ?)" + ) + params.append(tag) if folder: # Escape LIKE wildcards in the folder name diff --git a/prompt_search_list.py b/prompt_search_list.py index 6910d56..09a9319 100644 --- a/prompt_search_list.py +++ b/prompt_search_list.py @@ -5,6 +5,7 @@ OUTPUT_IS_LIST=True, allowing direct connection to nodes that accept list inputs. """ +import re from typing import Any, Dict, List, Tuple try: @@ -79,7 +80,7 @@ def INPUT_TYPES(cls) -> InputTypeDict: IO.STRING, { "default": "", - "tooltip": "Comma-separated list of tags to filter by", + "tooltip": "Comma-separated list of tags to filter by (partial match)", }, ), "min_rating": ( @@ -100,20 +101,33 @@ def INPUT_TYPES(cls) -> InputTypeDict: "tooltip": "Maximum number of results to return", }, ), + "skip_multipart": ( + "BOOLEAN", + { + "default": True, + "tooltip": "Skip prompts containing Clip_1/Clip_2/etc. multi-part markers", + }, + ), }, } - RETURN_TYPES = (IO.STRING,) - RETURN_NAMES = ("prompts",) - OUTPUT_IS_LIST = (True,) - OUTPUT_TOOLTIPS = ("List of prompt texts matching the search criteria.",) + RETURN_TYPES = (IO.STRING, IO.STRING, IO.INT) + RETURN_NAMES = ("prompts", "preview", "count") + OUTPUT_IS_LIST = (True, False, False) + OUTPUT_TOOLTIPS = ( + "List of prompt texts matching the search criteria.", + "Preview of all found prompts (for display).", + "Number of prompts found.", + ) FUNCTION = "search" OUTPUT_NODE = True CATEGORY = "🫶 ComfyAssets/🧠 Prompts" DESCRIPTION = ( "Searches the prompt database and outputs matching prompts as a list. " "Use this node to retrieve stored prompts for batch processing workflows. " - "Connect to nodes that accept list inputs like String OutputList or batch processors." + "'prompts' output sends each prompt individually to downstream nodes (batch). " + "'preview' output shows all found prompts as one text block. " + "'count' output shows how many prompts were found." ) def search( @@ -123,7 +137,8 @@ def search( tags: str = "", min_rating: int = 0, limit: int = 50, - ) -> Tuple[List[str]]: + skip_multipart: bool = True, + ): """ Search for prompts matching the given criteria. @@ -135,7 +150,7 @@ def search( limit: Maximum number of results to return Returns: - Tuple containing a list of prompt text strings + Dict with 'ui' display info and 'result' tuple """ try: # Parse tags from comma-separated string @@ -143,38 +158,69 @@ def search( if tags and tags.strip(): tags_list = [tag.strip() for tag in tags.split(",") if tag.strip()] - # Perform search + # Perform search with partial tag matching for discovery results = self.db.search_prompts( text=text.strip() if text and text.strip() else None, category=category.strip() if category and category.strip() else None, tags=tags_list, + tag_partial=True, rating_min=min_rating if min_rating > 0 else None, limit=limit, ) - # Extract just the prompt text from results - prompt_texts = [r["text"] for r in results if r.get("text")] - - self.logger.debug( - f"Search returned {len(prompt_texts)} prompts " - f"(text='{text[:20]}...' if text else '', category='{category}', " - f"tags={tags_list}, min_rating={min_rating}, limit={limit})" - ) - - # Return empty list if no results (not an error) + # Extract prompt text, collapsing newlines to spaces so downstream + # nodes that split on \n (e.g. StringOutputList) treat each DB + # entry as a single prompt. + prompt_texts = [ + " ".join(r["text"].split()) for r in results if r.get("text") + ] + + # Filter out multi-part prompts (Clip_1/Clip_2 markers) + if skip_multipart: + prompt_texts = [ + p for p in prompt_texts if not re.search(r"Clip_\d+", p) + ] + + # Filter out prompts that are only LoRA tags with no actual content. + # Uses subtraction instead of repeated groups to avoid backtracking. + prompt_texts = [ + p for p in prompt_texts if re.sub(r"]+>", "", p).strip() + ] + + count = len(prompt_texts) + + # Build a preview: numbered list of truncated prompts + preview_lines = [] + for i, p in enumerate(prompt_texts, 1): + truncated = p[:120] + "..." if len(p) > 120 else p + preview_lines.append(f"[{i}] {truncated}") + preview = "\n".join(preview_lines) if preview_lines else "No results found" + + # Must return at least one element — ComfyUI's slice_dict + # indexes into OUTPUT_IS_LIST outputs and crashes on empty lists. if not prompt_texts: self.logger.info("Search returned no results") - return ([],) + return { + "ui": {"text": ["No results found"]}, + "result": ([""], preview, count), + } - return (prompt_texts,) + return { + "ui": {"text": [f"Found {count} prompts"]}, + "result": (prompt_texts, preview, count), + } except Exception as e: self.logger.error(f"Search error: {e}", exc_info=True) - # Return empty list on error to avoid breaking workflows - return ([],) + return { + "ui": {"text": [f"Search error: {e}"]}, + "result": ([""], f"Error: {e}", 0), + } @classmethod - def IS_CHANGED(cls, text="", category="", tags="", min_rating=0, limit=50): + def IS_CHANGED( + cls, text="", category="", tags="", min_rating=0, limit=50, skip_multipart=True + ): """ Always re-execute to get fresh results from the database. diff --git a/pyproject.toml b/pyproject.toml index 04cc3b7..4fc861b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,7 @@ [project] name = "promptmanager" description = "A powerful ComfyUI custom node that extends the standard text encoder with persistent prompt storage, advanced search capabilities, and an automatic image gallery system using SQLite." -version = "3.2.1" +version = "3.2.2" license = {file = "LICENSE"} dependencies = ["# Core dependencies for PromptManager", "# Note: Most dependencies are already included with ComfyUI", "# Already included with Python standard library:", "# - sqlite3", "# - hashlib", "# - json", "# - datetime", "# - os", "# - typing", "# - threading", "# - uuid", "# Required for gallery functionality:", "watchdog>=2.1.0 # For file system monitoring", "Pillow>=8.0.0 # For image metadata extraction (usually included with ComfyUI)", "# Optional dependencies for enhanced search functionality:", "# fuzzywuzzy[speedup]>=0.18.0 # For fuzzy string matching (optional)", "# sqlalchemy>=1.4.0 # For advanced ORM features (optional)", "# Development dependencies (optional):", "# pytest>=6.0.0 # For running tests", "# black>=22.0.0 # For code formatting", "# flake8>=4.0.0 # For linting", "# mypy>=0.910 # For type checking"] diff --git a/tests/test_prompt_tracker.py b/tests/test_prompt_tracker.py index 2b8c6b2..a9bfdc8 100644 --- a/tests/test_prompt_tracker.py +++ b/tests/test_prompt_tracker.py @@ -34,6 +34,8 @@ def _make_tracker(prompt_timeout=600, cleanup_interval=300): tracker._local = threading.local() tracker.active_prompts = {} tracker.lock = threading.Lock() + tracker._prompt_queue = [] + tracker._queue_lock = threading.Lock() tracker.cleanup_interval = cleanup_interval tracker.prompt_timeout = prompt_timeout # Don't start cleanup thread in tests @@ -256,5 +258,59 @@ def test_expired_prompts_cleaned(self): self.assertEqual(len(expired_ids), 1) +class TestPromptQueue(unittest.TestCase): + """Test the FIFO prompt queue for batch workflow image linking.""" + + def setUp(self): + self.tracker = _make_tracker() + + def test_queue_empty_initially(self): + result = self.tracker.pop_next_prompt() + self.assertIsNone(result) + + def test_set_prompt_pushes_to_queue(self): + self.tracker.set_current_prompt("p1", {"prompt_id": 10}) + self.assertEqual(len(self.tracker._prompt_queue), 1) + + def test_pop_returns_fifo_order(self): + self.tracker.set_current_prompt("first", {"prompt_id": 1}) + self.tracker.set_current_prompt("second", {"prompt_id": 2}) + self.tracker.set_current_prompt("third", {"prompt_id": 3}) + + p1 = self.tracker.pop_next_prompt() + p2 = self.tracker.pop_next_prompt() + p3 = self.tracker.pop_next_prompt() + + self.assertEqual(p1["id"], 1) + self.assertEqual(p2["id"], 2) + self.assertEqual(p3["id"], 3) + + def test_pop_returns_none_when_exhausted(self): + self.tracker.set_current_prompt("only", {"prompt_id": 1}) + self.tracker.pop_next_prompt() + result = self.tracker.pop_next_prompt() + self.assertIsNone(result) + + def test_queue_preserves_prompt_text(self): + self.tracker.set_current_prompt("hello world", {"prompt_id": 5}) + ctx = self.tracker.pop_next_prompt() + self.assertEqual(ctx["text"], "hello world") + + def test_batch_simulation(self): + """Simulate a batch workflow: encode 10 prompts, then pop 10 in order.""" + ids = list(range(100, 110)) + for pid in ids: + self.tracker.set_current_prompt(f"prompt_{pid}", {"prompt_id": pid}) + + popped_ids = [] + while True: + ctx = self.tracker.pop_next_prompt() + if ctx is None: + break + popped_ids.append(ctx["id"]) + + self.assertEqual(popped_ids, ids) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_search_filters.py b/tests/test_search_filters.py new file mode 100644 index 0000000..d6ff2c1 --- /dev/null +++ b/tests/test_search_filters.py @@ -0,0 +1,135 @@ +""" +Tests for PromptSearchList output filtering. + +Tests the newline collapsing, multi-part (Clip_) filtering, +and LoRA-only filtering applied to search results. +""" + +import os +import re +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + + +# --- Helpers: replicate the filter logic from prompt_search_list.py --- + + +def apply_filters(texts, skip_multipart=True): + """Apply the same filtering chain as PromptSearchList.search().""" + # Collapse newlines + result = [" ".join(t.split()) for t in texts if t] + + # Filter multi-part + if skip_multipart: + result = [p for p in result if not re.search(r"Clip_\d+", p)] + + # Filter LoRA-only (subtraction approach — no backtracking risk) + result = [p for p in result if re.sub(r"]+>", "", p).strip()] + + return result + + +class TestNewlineCollapse(unittest.TestCase): + """Newlines must be collapsed so StringOutputList treats each DB entry as one prompt.""" + + def test_single_line_unchanged(self): + result = apply_filters(["a beautiful landscape"]) + self.assertEqual(result, ["a beautiful landscape"]) + + def test_multiline_collapsed_to_spaces(self): + result = apply_filters(["line one\nline two\nline three"]) + self.assertEqual(result, ["line one line two line three"]) + + def test_tabs_and_extra_spaces_collapsed(self): + result = apply_filters(["word1 \t word2\n\nword3"]) + self.assertEqual(result, ["word1 word2 word3"]) + + def test_leading_trailing_whitespace_stripped(self): + result = apply_filters([" padded prompt \n"]) + self.assertEqual(result, ["padded prompt"]) + + +class TestMultipartFilter(unittest.TestCase): + """Prompts containing Clip_N markers should be filtered when skip_multipart=True.""" + + def test_clip_markers_filtered(self): + texts = [ + "Clip_1 a video prompt Clip_2 continuation", + "a normal prompt", + ] + result = apply_filters(texts, skip_multipart=True) + self.assertEqual(result, ["a normal prompt"]) + + def test_clip_markers_kept_when_disabled(self): + texts = ["Clip_1 a video prompt Clip_2 continuation"] + result = apply_filters(texts, skip_multipart=False) + self.assertEqual(len(result), 1) + + def test_clip_in_word_filtered(self): + # "Clip_3" anywhere in text should match + texts = ["some text with Clip_3 marker"] + result = apply_filters(texts, skip_multipart=True) + self.assertEqual(result, []) + + def test_clip_without_number_not_filtered(self): + texts = ["clip art style painting"] + result = apply_filters(texts, skip_multipart=True) + self.assertEqual(result, ["clip art style painting"]) + + +class TestLoraOnlyFilter(unittest.TestCase): + """Prompts that are only LoRA tags with no actual content should be filtered.""" + + def test_single_lora_tag_filtered(self): + result = apply_filters([""]) + self.assertEqual(result, []) + + def test_multiple_lora_tags_filtered(self): + result = apply_filters([" "]) + self.assertEqual(result, []) + + def test_lora_with_content_kept(self): + result = apply_filters([" a beautiful painting"]) + self.assertEqual(result, [" a beautiful painting"]) + + def test_content_with_lora_in_middle_kept(self): + result = apply_filters(["masterpiece, , best quality"]) + self.assertEqual(result, ["masterpiece, , best quality"]) + + def test_lora_with_newlines_collapsed_then_filtered(self): + # After collapsing newlines, this becomes a single line of LoRA tags + result = apply_filters(["\n"]) + self.assertEqual(result, []) + + def test_empty_string_filtered(self): + result = apply_filters([""]) + self.assertEqual(result, []) + + +class TestFilterCombination(unittest.TestCase): + """Test all filters working together on a mixed batch.""" + + def test_mixed_batch(self): + texts = [ + "a good prompt", + "", + "Clip_1\nfirst segment\nClip_2\nsecond segment", + " nice painting, flowers", + "another\ngood\nprompt", + " ", + ] + result = apply_filters(texts) + self.assertEqual( + result, + [ + "a good prompt", + " nice painting, flowers", + "another good prompt", + ], + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/utils/image_monitor.py b/utils/image_monitor.py index b49ba77..892cec8 100644 --- a/utils/image_monitor.py +++ b/utils/image_monitor.py @@ -80,14 +80,26 @@ def on_created(self, event): directory. It filters for image files and schedules them for processing after a small delay to ensure the file is fully written. + The prompt context is captured immediately (before the delay) to handle + batch workflows where the prompt tracker advances before images are processed. + Args: event: FileSystemEvent object containing event details """ if not event.is_directory and self.is_image_file(event.src_path): self.logger.info(f"New image detected: {event.src_path}") - # Small delay to ensure file is fully written + # Snapshot prompt context NOW before the delay — in batch workflows + # the tracker advances to the next prompt before images are processed. + prompt_snapshot = self.prompt_tracker.get_current_prompt() + if prompt_snapshot: + self.logger.debug( + f"Snapshot prompt {prompt_snapshot.get('id', '?')} for {os.path.basename(event.src_path)}" + ) threading.Timer( - self.processing_delay, self.process_new_image, args=[event.src_path] + self.processing_delay, + self.process_new_image, + args=[event.src_path], + kwargs={"prompt_snapshot": prompt_snapshot}, ).start() def is_image_file(self, filepath: str) -> bool: @@ -105,18 +117,19 @@ def is_image_file(self, filepath: str) -> bool: return False return filepath.lower().endswith(self.supported_extensions) - def process_new_image(self, image_path: str): + def process_new_image(self, image_path: str, prompt_snapshot=None): """Process a newly created image file for gallery integration. This method handles the complete processing pipeline for a new image: 1. Verifies the file still exists - 2. Gets the current prompt context from the tracker - 3. Extracts ComfyUI metadata from the image - 4. Links the image to the appropriate prompt in the database - 5. Handles fallback scenarios when no active prompt is available + 2. Extracts ComfyUI metadata from the image + 3. Tries to identify the correct prompt via metadata-based lookup + 4. Falls back to the prompt snapshot captured at event time + 5. Links the image to the appropriate prompt in the database Args: image_path: Full path to the newly created image file + prompt_snapshot: Prompt context captured at file-creation time (optional) """ try: self.logger.info(f"Processing image: {image_path}") @@ -125,35 +138,61 @@ def process_new_image(self, image_path: str): self.logger.warning(f"Image file no longer exists: {image_path}") return - # Get current prompt context first - current_prompt = self.prompt_tracker.get_current_prompt() - self.logger.info( - f"Current prompt context: {current_prompt['id'] if current_prompt else 'None'}" - ) + # Extract ComfyUI metadata first — needed both for linking and prompt lookup + metadata = None + try: + metadata = self.metadata_extractor.extract_metadata(image_path) + self.logger.debug(f"Extracted metadata: {bool(metadata)}") + except Exception as meta_error: + self.logger.warning(f"Metadata extraction failed: {meta_error}") + # Strategy 1: Pop from batch queue (most reliable for batch workflows). + # Prompts are queued during CLIP encoding in order; images save in + # the same order, so FIFO pop gives the correct prompt per image. + current_prompt = self.prompt_tracker.pop_next_prompt() + if current_prompt: + self.logger.info( + f"Queue match: prompt {current_prompt['id']} for " + f"{os.path.basename(image_path)}" + ) + + # Strategy 2: Find prompt from image metadata + if not current_prompt: + current_prompt = self._find_prompt_from_metadata(metadata) + if current_prompt: + self.logger.info(f"Metadata match: prompt {current_prompt['id']}") + + # Strategy 3: Use snapshot captured at file-creation time + if not current_prompt and prompt_snapshot: + current_prompt = prompt_snapshot + self.logger.info( + f"Snapshot match: prompt {current_prompt.get('id', 'unknown')}" + ) + + # Strategy 4: Check live prompt tracker + if not current_prompt: + current_prompt = self.prompt_tracker.get_current_prompt() + if current_prompt: + self.logger.info( + f"Live tracker match: prompt {current_prompt['id']}" + ) + + # Strategy 5: Fallback to most recent prompt in DB if not current_prompt: - self.logger.debug(f"No active prompt context for image: {image_path}") - # Fallback: try to link to the most recent prompt in database current_prompt = self._get_fallback_prompt() if current_prompt: - self.logger.debug(f"Using fallback prompt: {current_prompt['id']}") + self.logger.debug(f"Fallback match: prompt {current_prompt['id']}") else: - self.logger.warning(f"No fallback prompt available, skipping image") + self.logger.warning( + f"No prompt context available, skipping image: {image_path}" + ) return - else: - # Extend the timeout for this prompt since we're still getting images - if "execution_id" in current_prompt: - self.prompt_tracker.extend_prompt_timeout( - current_prompt["execution_id"], 300 - ) # Add 5 more minutes - # Extract ComfyUI metadata - try: - metadata = self.metadata_extractor.extract_metadata(image_path) - self.logger.debug(f"Extracted metadata: {bool(metadata)}") - except Exception as meta_error: - self.logger.warning(f"Metadata extraction failed: {meta_error}") - metadata = None + # Extend timeout if we have an active execution + if current_prompt.get("execution_id"): + self.prompt_tracker.extend_prompt_timeout( + current_prompt["execution_id"], 300 + ) if metadata: self.logger.debug( @@ -164,7 +203,6 @@ def process_new_image(self, image_path: str): self.logger.debug( f"Linking image with basic info to prompt {current_prompt['id']}" ) - # Link with basic file info even without metadata basic_metadata = self.get_basic_file_info(image_path) self.link_image_to_prompt( image_path, current_prompt, {"file_info": basic_metadata} @@ -176,6 +214,87 @@ def process_new_image(self, image_path: str): self.logger.error(traceback.format_exc()) + def _find_prompt_from_metadata(self, metadata): + """Extract prompt text from image metadata and look up the matching DB prompt. + + Parses the ComfyUI workflow/prompt data embedded in the image to find + PromptManager node inputs, then matches against the database by hash. + + Args: + metadata: Extracted metadata dict from the image, or None + + Returns: + Prompt context dict with 'id' and 'text', or None if not found + """ + if not metadata: + return None + + prompt_text = None + + # Try to find PromptManager node in the prompt execution data + prompt_data = metadata.get("prompt") + if isinstance(prompt_data, dict): + for node_id, node_info in prompt_data.items(): + if not isinstance(node_info, dict): + continue + class_type = node_info.get("class_type", "") + if class_type in ("PromptManager", "PromptManagerText"): + inputs = node_info.get("inputs", {}) + text = inputs.get("text", "") + if text and isinstance(text, str) and text.strip(): + prompt_text = text.strip() + break + + # Fallback: check text_encoder_nodes from workflow, but only if + # the text input is NOT connected (connected inputs override widget values, + # so the widget value would be stale in batch workflows). + if not prompt_text: + text_nodes = metadata.get("text_encoder_nodes", []) + for node in text_nodes: + node_type = node.get("type") or node.get("class_type") or "" + if "PromptManager" in node_type: + # Check if text input is connected — if so, widget value is stale + text_connected = False + for inp in node.get("inputs", []): + if isinstance(inp, dict) and inp.get("name") == "text": + if inp.get("link") is not None: + text_connected = True + break + if text_connected: + self.logger.debug( + "PromptManager text input is connected — " + "skipping stale widget value, deferring to snapshot" + ) + continue + widgets = node.get("widgets_values", []) + if widgets and isinstance(widgets[0], str) and widgets[0].strip(): + prompt_text = widgets[0].strip() + break + + if not prompt_text: + return None + + # Look up by hash in database + try: + import hashlib + + normalized = prompt_text.strip().lower() + prompt_hash = hashlib.sha256(normalized.encode("utf-8")).hexdigest() + existing = self.db_manager.get_prompt_by_hash(prompt_hash) + if existing: + self.logger.debug( + f"Found DB prompt {existing['id']} from metadata text" + ) + return { + "id": existing["id"], + "text": existing["text"], + "from_metadata": True, + } + except Exception as e: + self.logger.warning(f"Metadata-based prompt lookup failed: {e}") + + return None + def get_basic_file_info(self, image_path: str) -> Dict[str, Any]: """Get basic file information when metadata extraction fails. diff --git a/utils/prompt_tracker.py b/utils/prompt_tracker.py index 183acc5..2c32298 100644 --- a/utils/prompt_tracker.py +++ b/utils/prompt_tracker.py @@ -76,6 +76,10 @@ def __init__(self, db_manager): self._local = threading.local() self.active_prompts = {} # Global tracking for multiple threads self.lock = threading.Lock() + # FIFO queue for batch workflows: prompt_ids are pushed during encoding + # and popped during image linking, preserving correct ordering. + self._prompt_queue = [] + self._queue_lock = threading.Lock() # Read from GalleryConfig if available, otherwise use defaults try: @@ -159,6 +163,13 @@ def set_current_prompt( with self.lock: self.active_prompts[execution_id] = execution_context + # Push to batch queue for ordered image linking + with self._queue_lock: + self._prompt_queue.append(execution_context) + self.logger.debug( + f"Queue size: {len(self._prompt_queue)} after push for prompt {prompt_id}" + ) + self.logger.debug( f"Set current prompt: {execution_id} -> {prompt_text[:50]}... (thread: {threading.current_thread().ident})" ) @@ -207,6 +218,32 @@ def get_current_prompt(self) -> Optional[Dict[str, Any]]: ) return None + def pop_next_prompt(self) -> Optional[Dict[str, Any]]: + """Pop the next prompt from the batch queue (FIFO). + + In batch workflows, prompts are encoded in order and images are saved + in the same order. This method pops the oldest queued prompt, ensuring + each image links to the correct prompt by position. + + Skips and discards expired entries to prevent stale prompts from + accumulating across sessions. + + Returns: + Prompt context dict, or None if queue is empty or all entries expired + """ + now = time.time() + with self._queue_lock: + while self._prompt_queue: + ctx = self._prompt_queue.pop(0) + if now - ctx.get("timestamp", 0) < self.prompt_timeout: + self.logger.debug( + f"Queue pop: prompt {ctx.get('id')} " + f"({len(self._prompt_queue)} remaining)" + ) + return ctx + self.logger.debug(f"Queue discard expired: prompt {ctx.get('id')}") + return None + def _find_recent_prompt(self) -> Optional[Dict[str, Any]]: """Find the most recent prompt that's still valid. @@ -335,6 +372,18 @@ def _cleanup_expired_prompts(self): if expired_ids: self.logger.debug(f"Cleaned up {len(expired_ids)} expired prompts") + # Also prune expired entries from the batch queue + with self._queue_lock: + before = len(self._prompt_queue) + self._prompt_queue = [ + ctx + for ctx in self._prompt_queue + if current_time - ctx.get("timestamp", 0) < self.prompt_timeout + ] + pruned = before - len(self._prompt_queue) + if pruned: + self.logger.debug(f"Pruned {pruned} expired queue entries") + time.sleep(self.cleanup_interval) except Exception as e: