🔍 Code Extractor

Comprehensive test function that validates the auto-continuation workflow for multi-turn conversations, including session detection from PDFs, conversation context preservation, and automatic conversation switching.

Purpose

This async test function validates the complete auto-continuation workflow system by: (1) checking for existing session PDFs from previous conversations, (2) testing session detection capabilities on those PDFs, (3) creating a test image for continuation testing, (4) demonstrating manual and automatic continuation workflows, and (5) showing how session information is embedded and detected in PDF responses. It serves as both a test suite and documentation for the auto-continuation feature.

Source Code

async def test_auto_continuation_workflow():
    """Test the complete auto-continuation workflow"""
    print("🎯 TESTING AUTO-CONTINUATION WORKFLOW")
    print("=" * 70)
    
    # Step 1: Verify we have existing session PDFs
    print("\n📋 Step 1: Checking for existing session PDFs...")
    
    existing_pdfs = [
        "../RESPONSE_conv_20250731_224420_6a63a783_ex001_test_session_sample.pdf",
        "ERROR_conv_20250731_224420_6a63a783_ex002_test_followup.pdf"
    ]
    
    available_pdf = None
    for pdf_path in existing_pdfs:
        if Path(pdf_path).exists():
            available_pdf = pdf_path
            print(f"   ✅ Found: {Path(pdf_path).name}")
            break
    
    if not available_pdf:
        print("   ❌ No existing session PDFs found for testing")
        print("   💡 Run a file processing test first to generate session PDFs")
        return
    
    # Step 2: Test session detection on existing PDF
    print(f"\n🔍 Step 2: Testing session detection on existing PDF...")
    from session_detector import detect_session_from_file
    
    detected = detect_session_from_file(available_pdf)
    if detected:
        print(f"   ✅ Session detected successfully!")
        print(f"      🆔 Conversation: {detected.conversation_id}")
        print(f"      #️⃣  Exchange: {detected.exchange_number}")
        print(f"      🎯 Confidence: {detected.confidence:.2f}")
        print(f"      📍 Source: {detected.source}")
    else:
        print("   ❌ Failed to detect session from PDF")
        return
    
    # Step 3: Create a simple image to test continuation
    print(f"\n🖼️  Step 3: Creating test image for continuation...")
    
    test_image_path = Path("test_continuation.png")
    
    # Create a simple test image using ImageMagick
    import subprocess
    try:
        subprocess.run([
            "convert", "-size", "400x300", "xc:white",
            "-pointsize", "16", "-fill", "black",
            "-annotate", "+20+50", 
            f"Follow-up Question\\n\\nThis should continue\\nconversation:\\n{detected.conversation_id}\\n\\nFrom exchange #{detected.exchange_number}\\n\\nPlease provide additional\\ninformation or clarification.",
            str(test_image_path)
        ], check=True, capture_output=True)
        print(f"   ✅ Test image created: {test_image_path}")
    except subprocess.CalledProcessError as e:
        print(f"   ❌ Failed to create test image: {e}")
        print("   💡 ImageMagick (convert) not available, creating placeholder")
        # Create a placeholder file
        test_image_path.write_text("Test continuation image placeholder")
    
    # Step 4: Test auto-continuation with explicit conversation ID
    print(f"\n🚀 Step 4: Testing manual continuation (for comparison)...")
    print(f"   Command: python main.py --file {test_image_path} --conversation-id {detected.conversation_id}")
    
    # Step 5: Test auto-continuation with auto-detection
    print(f"\n🤖 Step 5: Auto-detection would work as follows:")
    print(f"   1. User drops PDF response back into watch folder")
    print(f"   2. System detects session: {detected.conversation_id}")
    print(f"   3. System automatically switches to that conversation")
    print(f"   4. System continues from exchange #{detected.exchange_number + 1}")
    print(f"   5. Response maintains conversation context and history")
    
    # Step 6: Show the workflow
    print(f"\n💡 WORKFLOW DEMONSTRATION:")
    print(f"   Original conversation: {detected.conversation_id}")
    print(f"   Last exchange: #{detected.exchange_number}")
    print(f"   📄 PDF contains session info in:")
    print(f"      • Filename: {Path(available_pdf).name}")
    print(f"      • PDF metadata (title, subject, creator)")
    print(f"      • Footer: 'Session: {detected.conversation_id} | Exchange #{detected.exchange_number}'")
    print(f"      • Content text (if needed)")
    print(f"   🔄 Auto-continuation:")
    print(f"      • Detects session with {detected.confidence:.0%} confidence")
    print(f"      • Switches to conversation {detected.conversation_id}")
    print(f"      • Starts new exchange #{detected.exchange_number + 1}")
    print(f"      • Includes conversation history as context")
    
    # Cleanup
    if test_image_path.exists() and test_image_path.stat().st_size > 100:
        print(f"\n🧹 Cleanup: Removing test image")
        test_image_path.unlink()
    
    print(f"\n🎉 AUTO-CONTINUATION WORKFLOW TEST COMPLETE!")
    print(f"✅ Session detection: WORKING")
    print(f"✅ Auto-conversation switching: IMPLEMENTED")
    print(f"✅ Context preservation: READY")
    print(f"✅ PDF session tracking: ACTIVE")

Return Value

Returns None. This is a test function that outputs results to stdout, demonstrating the workflow through print statements. It performs validation checks and displays success/failure status for each step of the auto-continuation process.

Dependencies

• asyncio
• tempfile
• shutil
• pathlib
• subprocess
• session_detector

Required Imports

import asyncio
import tempfile
import shutil
from pathlib import Path

Conditional/Optional Imports

These imports are only needed under specific conditions:

from session_detector import detect_session_from_file

import subprocess

Usage Example

import asyncio

async def main():
    # Run the auto-continuation workflow test
    await test_auto_continuation_workflow()
    # This will:
    # 1. Check for existing session PDFs
    # 2. Detect session information from PDFs
    # 3. Create a test continuation image
    # 4. Demonstrate manual and auto-continuation workflows
    # 5. Show session tracking and context preservation

# Execute the test
asyncio.run(main())

# Or run directly if already in async context:
# await test_auto_continuation_workflow()

Best Practices

• Run file processing tests first to generate session PDFs before running this test
• Ensure session_detector module is properly installed and accessible
• The function expects specific PDF naming conventions: RESPONSE_conv_* or ERROR_conv_*
• ImageMagick is optional but recommended for full test coverage
• This is a demonstration/test function - review output carefully to understand the workflow
• The function creates temporary files (test_continuation.png) which are cleaned up automatically
• Session PDFs must contain embedded session information in filename, metadata, and footer
• The test validates confidence scores for session detection (should be high for reliable detection)
• Use this test to verify the complete round-trip workflow: initial request → PDF response → continuation request
• The function is non-destructive and safe to run multiple times

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function test_session_detection 61.0% similar

  A comprehensive test function that validates session detection capabilities from multiple sources including filenames, PDF files, and text patterns.

  From: /tf/active/vicechatdev/e-ink-llm/test_session_detection.py

• function test_workflow_progress_structure 60.3% similar

  A test function that validates the structure and behavior of a workflow progress tracking system for SQL query processing, including progress states, step transitions, and completion data.

  From: /tf/active/vicechatdev/full_smartstat/test_enhanced_progress.py

• function test_enhanced_workflow 60.1% similar

  A comprehensive test function that validates the EnhancedSQLWorkflow system by testing component initialization, request parsing, and data assessment capabilities.

  From: /tf/active/vicechatdev/full_smartstat/test_enhanced_workflow.py

• function test_session_manager 59.7% similar

  A comprehensive test function that validates the SessionManager class functionality including conversation creation, exchange tracking, filename generation, and context retrieval.

  From: /tf/active/vicechatdev/e-ink-llm/test_improvements.py

• function test_pdf_session_integration 59.1% similar

  Integration test function that verifies PDF generation includes session tracking information by creating both response and error PDFs with conversation IDs and exchange numbers.

  From: /tf/active/vicechatdev/e-ink-llm/test_pdf_session.py