Function test_soft_interrupt_with_tools

scripts/test_soft_interrupt.py:284–373

Test soft interrupt injection when tools are involved. 1. Send message that triggers a tool 2. Queue interrupt during tool execution 3. Verify interrupt appears after tool result

()

        destroy_session(session_id)

def test_soft_interrupt_with_tools():
    """
    Test soft interrupt injection when tools are involved.
    
    1. Send message that triggers a tool
    2. Queue interrupt during tool execution
    3. Verify interrupt appears after tool result
    """
    print("\n" + "="*60)
    print("TEST: Soft interrupt with tool execution")
    print("="*60)
    
    session_id = create_test_session(cwd="/tmp")
    if not session_id:
        print("❌ Failed to create session")
        return False
    print(f"Created session: {session_id}")
    
    try:
        # Create a test file
        test_file = "/tmp/test_interrupt_tools.txt"
        with open(test_file, 'w') as f:
            f.write("apple\nbanana\ncherry\ndate\nelderberry")
        
        # Start message that will trigger file read
        result_q = queue_mod.Queue()
        thread = threading.Thread(
            target=send_message_async,
            args=(f"Read {test_file} and list each fruit.", session_id, result_q)
        )
        thread.start()
        
        # Wait for tool execution to start, then queue interrupt
        time.sleep(0.5)
        print("Queueing soft interrupt: 'How many fruits are there?'")
        queue_interrupt(session_id, "How many fruits are there in total?")
        
        # Wait for completion
        thread.join(timeout=180)
        
        if thread.is_alive():
            print("❌ Timed out")
            os.remove(test_file)
            return False
        
        ok, output = result_q.get()
        os.remove(test_file)
        
        if not ok:
            print(f"❌ Failed: {output[:200]}")
            return False
        
        print(f"Response: {output[:150]}...")
        
        # Check history
        history = get_history(session_id)
        print(f"\nHistory ({len(history)} messages):")
        print_history(history)