Function test_interrupt_during_long_response

scripts/test_soft_interrupt.py:456–540

Test soft interrupt during a genuinely long response. We ask for something that takes time to generate.

()

        destroy_session(session_id)

def test_interrupt_during_long_response():
    """
    Test soft interrupt during a genuinely long response.
    We ask for something that takes time to generate.
    """
    print("\n" + "="*60)
    print("TEST: Interrupt during long response")
    print("="*60)
    
    session_id = create_test_session()
    if not session_id:
        print("❌ Failed to create session")
        return False
    print(f"Created session: {session_id}")
    
    try:
        # Ask for something that takes time
        result_q = queue_mod.Queue()
        thread = threading.Thread(
            target=send_message_async,
            args=("Write a detailed 5-paragraph essay about the history of computing, from ENIAC to modern smartphones.", session_id, result_q)
        )
        thread.start()
        
        # Queue interrupt after a delay
        time.sleep(2.0)
        print("Queueing interrupt: 'STOP - just say OK'")
        ok = queue_interrupt(session_id, "STOP - just say 'OK' and nothing else.")
        if ok:
            print("✓ Interrupt queued")
        
        # Wait for completion
        thread.join(timeout=180)
        
        if thread.is_alive():
            print("❌ Timed out")
            return False
        
        ok, _ = result_q.get()
        if not ok:
            print("❌ Message failed")
            return False
        
        # Check history
        history = get_history(session_id)
        print(f"\nHistory ({len(history)} messages):")
        print_history(history)
        
        # Look for our interrupt
        found_stop = False
        for msg in history:
            if msg.get('role') == 'user':
                text = extract_text_from_content(msg.get('content', []))
                if 'STOP' in text:
                    found_stop = True
                    break
        
        if found_stop: