server/tests/test_questions.py
Rama main · réplica de lectura del repositorio original.
"""Ask-user behavior through the real SDK runner and application controller."""
import asyncio
import json
from copy import deepcopy
import pytest
import pytest_asyncio
from fastapi.testclient import TestClient
from bumblehive import BumblehiveRuntime
from bumblehive.providers.base import ModelProvider, ModelResponse
from bumblehive.protocols import ToolCall
from bumblehive.session.manager import SessionManager
from bumblehive_server.app import create_app
from bumblehive_server.chat import questions
from bumblehive_server.chat.session import InputError
from bumblehive_server.runtime_service import RuntimeService
from bumblehive_server.schemas import ChatRequest
from bumblehive_server.session_reader import SessionReader
class QuestionProvider(ModelProvider):
def __init__(self):
self.requests = []
self.after_tool = asyncio.Event()
self.release = asyncio.Event()
self.release.set()
async def generate(self, request):
self.requests.append(deepcopy(request.messages))
tools = [m for m in request.messages if m["role"] == "tool"]
if not tools:
return ModelResponse(None, [ToolCall("ask", "ask_user", {
"question": "Which database?",
"options": ["PostgreSQL (recommended): flexible queries", "MySQL"],
})], "tool_calls")
self.after_tool.set()
await self.release.wait()
return ModelResponse(content="Question handled")
async def generate_stream(self, request, *, callbacks=None):
response = await self.generate(request)
if response.content and callbacks and callbacks.on_content_delta:
await callbacks.on_content_delta(response.content)
return response
def make_service(tmp_path):
reader = SessionReader(tmp_path / "sessions")
provider = QuestionProvider()
def factory(config):
runtime = BumblehiveRuntime(config)
runtime.sessions = SessionManager(reader.directory)
async def get_provider(_config):
return provider
runtime._get_provider = get_provider
return runtime
service = RuntimeService(
tmp_path / "config.json", session_reader=reader, runtime_factory=factory,
)
return service, provider, reader
@pytest_asyncio.fixture
async def asking(tmp_path, monkeypatch):
monkeypatch.setattr(questions, "ASK_USER_WAIT_SECONDS", 0.05)
service, provider, reader = make_service(tmp_path)
await service.startup()
try:
session_id = await reader.create(tmp_path)
controller = service.get_session_controller(session_id)
queue, _ = controller.subscribe()
yield service, provider, reader, controller, queue, tmp_path
finally:
await service.shutdown()
assert service._active_runs == 0
async def receive(queue, kind):
async with asyncio.timeout(3):
while True:
item = await queue.get()
if item["type"] == kind:
return item
async def start(asking):
_, _, _, controller, queue, workspace = asking
await controller.submit(ChatRequest(
content="Ask me", approval_mode="request",
config={"runtime": {"workspace": str(workspace)}},
))
return await receive(queue, "question_request")
def tool_results(messages):
return [json.loads(m["content"]) for m in messages if m["role"] == "tool"]
def background_results(messages):
return [json.loads(m["content"].split("\n\n", 1)[1]) for m in messages
if m["role"] == "user" and str(m["content"]).startswith("[Background tool result]\n")]
def late_answers(messages):
return [result for result in background_results(messages) if result["status"] == "answered"]
@pytest.mark.asyncio
@pytest.mark.parametrize("timed_out", [False, True])
async def test_concurrent_sessions_keep_questions_and_answers_isolated(asking, monkeypatch, timed_out):
monkeypatch.setattr(questions, "ASK_USER_WAIT_SECONDS", 0.05 if timed_out else 10)
service, provider, reader, first, first_queue, workspace = asking
second = service.get_session_controller(await reader.create(workspace))
second_queue, _ = second.subscribe()
other = (service, provider, reader, second, second_queue, workspace)
first_card, second_card = await asyncio.gather(start(asking), start(other))
assert first_card["question"] == second_card["question"]
# Tool call IDs can repeat across sessions; each controller owns its question.
assert first_card["task_id"] == second_card["task_id"] == "ask"
if timed_out:
await asyncio.gather(receive(first_queue, "result"), receive(second_queue, "result"))
await asyncio.gather(*(c.active_task for c in (first, second) if c.active_task))
# A session without this task cannot answer or skip either question.
empty = service.get_session_controller(await reader.create(workspace))
with pytest.raises(InputError, match="does not exist"):
await empty.answer_question(first_card["task_id"], "wrong session")
with pytest.raises(InputError, match="does not exist"):
await empty.answer_question(second_card["task_id"])
assert [q["task_id"] for q in first.subscribe()[1]["questions"]] == [first_card["task_id"]]
assert [q["task_id"] for q in second.subscribe()[1]["questions"]] == [second_card["task_id"]]
await asyncio.gather(
first.answer_question(first_card["task_id"], "first-session-only"),
second.answer_question(second_card["task_id"]),
)
for controller, queue, card in ((first, first_queue, first_card), (second, second_queue, second_card)):
assert await receive(queue, "question_closed") == {
"type": "question_closed", "session_id": controller.session_id,
"task_id": card["task_id"],
}
assert controller.subscribe()[1]["questions"] == []
await receive(first_queue, "result")
await receive(second_queue, "result")
first_history = (await reader.get(first.session_id)).messages
second_history = (await reader.get(second.session_id)).messages
assert "first-session-only" in json.dumps(first_history)
assert "first-session-only" not in json.dumps(second_history)
assert "wrong session" not in json.dumps(first_history + second_history)
assert tool_results(first_history)[0]["status"] == ("pending" if timed_out else "answered")
assert tool_results(second_history)[0]["status"] == ("pending" if timed_out else "skipped")
if timed_out:
# Both model calls use "ask", but questions and answers remain session-owned.
assert tool_results(first_history)[0] == tool_results(second_history)[0] == {
"status": "pending", "task_id": "ask",
}
assert len(late_answers(first_history)) == int(timed_out)
assert late_answers(second_history) == []
@pytest.mark.asyncio
@pytest.mark.parametrize("answer", ["MySQL", "Use our existing SQLite database"])
async def test_timely_answer_is_only_a_tool_result(asking, answer):
service, _, reader, controller, queue, _ = asking
card = await start(asking)
tool = service._runtime.tools.get_tool("ask_user")
assert card["task_id"] == "ask"
assert tool.parallel_safe is False
assert tool.parameters["required"] == ["question"]
await controller.answer_question(card["task_id"], answer)
with pytest.raises(InputError, match="does not exist"):
await controller.answer_question(card["task_id"], "duplicate")
await receive(queue, "result")
history = (await reader.get(controller.session_id)).messages
assert tool_results(history) == [{
"task_id": card["task_id"], "status": "answered", "answer": answer,
}]
assert late_answers(history) == []
assert controller.subscribe()[1]["questions"] == []
assert controller.questions == {}
@pytest.mark.asyncio
@pytest.mark.parametrize("running", [False, True])
async def test_timeout_late_answer_routes_once_and_keeps_original_result(asking, running):
_, provider, reader, controller, queue, _ = asking
if running:
provider.release.clear()
card = await start(asking)
if running:
await asyncio.wait_for(provider.after_tool.wait(), 3)
else:
await receive(queue, "result")
if controller.active_task:
await controller.active_task
item = controller.questions[card["task_id"]]
assert item.future is not None and not item.future.done()
background_task = item.background_task
results = await asyncio.gather(
controller.answer_question(card["task_id"], "SQLite"),
controller.answer_question(card["task_id"], "SQLite"),
return_exceptions=True,
)
assert results[0] is None
assert isinstance(results[1], InputError)
await background_task
assert controller.questions == {}
provider.release.set()
async with asyncio.timeout(3):
while controller.active_task:
await asyncio.sleep(0.001)
history = (await reader.get(controller.session_id)).messages
assert tool_results(history)[0] == {"status": "pending", "task_id": "ask"}
assert card["task_id"] == "ask"
assert len(late_answers(history)) == 1
assert background_results(history) == [{"task_id": "ask", "status": "answered", "answer": "SQLite"}]
assert controller._last_request.config["runtime"]["workspace"] == str(asking[-1])
assert controller._last_request.approval_mode == "request"
@pytest.mark.asyncio
async def test_disconnect_snapshot_and_cancel_release_waiter_but_keep_question(asking, monkeypatch):
monkeypatch.setattr(questions, "ASK_USER_WAIT_SECONDS", 10)
_, _, reader, controller, queue, _ = asking
card = await start(asking)
controller.unsubscribe(queue)
queue, snapshot = controller.subscribe()
assert snapshot["questions"] == [{
"task_id": card["task_id"], "question": card["question"],
"options": card["options"],
}]
assert controller.active_task is not None
await controller.cancel()
assert controller.questions[card["task_id"]].future is None
assert controller.subscribe()[1]["questions"] == snapshot["questions"]
await controller.answer_question(card["task_id"], "MySQL")
await receive(queue, "result")
history = (await reader.get(controller.session_id)).messages
assert len(late_answers(history)) == 1
assert controller.subscribe()[1]["questions"] == []
await controller.close()
assert controller.questions == {}
@pytest.mark.asyncio
async def test_unknown_blank_and_closed_answers_do_not_lock_card(asking):
_, _, _, controller, _, _ = asking
card = await start(asking)
with pytest.raises(InputError, match="does not exist"):
await controller.answer_question("other-session-question", "answer")
with pytest.raises(InputError, match="blank"):
await controller.answer_question(card["task_id"], " ")
assert card["task_id"] in controller.questions
controller._closing = True
with pytest.raises(InputError, match="closed"):
await controller.answer_question(card["task_id"], "answer")
assert card["task_id"] in controller.questions
controller._closing = False
@pytest.mark.asyncio
async def test_failed_answer_after_foreground_stop_keeps_question_for_retry(asking, monkeypatch):
monkeypatch.setattr(questions, "ASK_USER_WAIT_SECONDS", 10)
_, _, reader, controller, queue, _ = asking
card = await start(asking)
await controller.cancel()
async def reject_submission(*args, **kwargs):
raise InputError("submission_failed", "could not accept answer")
with monkeypatch.context() as patch:
patch.setattr(controller, "_submit", reject_submission)
with pytest.raises(InputError, match="could not accept answer"):
await controller.answer_question(card["task_id"], "MySQL")
assert controller.subscribe()[1]["questions"][0]["task_id"] == card["task_id"]
await controller.answer_question(card["task_id"], "MySQL")
await receive(queue, "result")
assert controller.subscribe()[1]["questions"] == []
history = (await reader.get(controller.session_id)).messages
assert len(late_answers(history)) == 1
@pytest.mark.asyncio
async def test_answer_during_stop_is_queued_without_starting_or_resolving_waiter(asking):
_, _, _, controller, _, _ = asking
card = await start(asking)
# The cancel path sets this flag before waiting for the run to unwind.
controller._stopping = True
item = controller.questions[card["task_id"]]
await controller.answer_question(card["task_id"], "MySQL")
assert card["task_id"] not in controller.questions
assert not item.future.done()
assert [p.request.message_id for p in controller.pending] == [card["task_id"]]
await controller.cancel()
assert controller.active_task is None
assert len(controller.pending) == 1
@pytest.mark.asyncio
@pytest.mark.parametrize("answer_first", [True, False])
async def test_zero_timeout_and_answer_choose_exactly_one_route(asking, monkeypatch, answer_first):
monkeypatch.setattr(questions, "ASK_USER_WAIT_SECONDS", 0)
_, _, reader, controller, queue, _ = asking
card = await start(asking)
if not answer_first:
await receive(queue, "result")
if controller.active_task:
await controller.active_task
await controller.answer_question(card["task_id"], "MySQL")
await receive(queue, "question_closed")
async with asyncio.timeout(3):
while controller.active_task:
await asyncio.sleep(0.001)
history = (await reader.get(controller.session_id)).messages
result = tool_results(history)[0]
assert (result["status"] == "answered") + len(late_answers(history)) == 1
@pytest.mark.asyncio
async def test_tool_rejects_missing_session_without_creating_controller(asking):
service, _, _, _, _, _ = asking
before = set(service._session_controllers)
with pytest.raises(RuntimeError, match="active server conversation"):
await service._runtime.tools.get_tool("ask_user").execute(question="Which database?")
assert set(service._session_controllers) == before
def test_websocket_answers_errors_and_reconnect(tmp_path, monkeypatch):
monkeypatch.setattr(questions, "ASK_USER_WAIT_SECONDS", 10)
service, _, reader = make_service(tmp_path)
app = create_app(runtime_service=service, session_reader=reader)
with TestClient(app) as client:
with client.websocket_connect("/ws/v1/chat/questions-test") as socket:
assert socket.receive_json()["questions"] == []
socket.send_json({"type": "message", "content": "Ask", "config": {
"runtime": {"workspace": str(tmp_path)},
}})
while True:
frame = socket.receive_json()
if frame["type"] == "question_request":
break
task_id = frame["task_id"]
socket.send_json({"type": "question_answer", "task_id": task_id, "answer": " "})
error = socket.receive_json()
while error["type"] != "error":
error = socket.receive_json()
assert error["scope"] == "input"
assert error["task_id"] == task_id
with client.websocket_connect("/ws/v1/chat/questions-test") as socket:
ready = socket.receive_json()
assert ready["questions"][0]["task_id"] == task_id
assert set(ready["questions"][0]) == {"task_id", "question", "options"}
socket.send_json({"type": "question_answer", "task_id": task_id, "answer": "MySQL"})
answered = socket.receive_json()
assert answered == {"type": "question_closed", "session_id": "questions-test", "task_id": task_id}
while socket.receive_json()["type"] != "result":
pass
with client.websocket_connect("/ws/v1/chat/questions-test") as socket:
assert socket.receive_json()["questions"] == []
@pytest.mark.asyncio
async def test_skip_ends_wait_immediately_and_rejects_later_answers(asking, monkeypatch):
monkeypatch.setattr(questions, "ASK_USER_WAIT_SECONDS", 60)
_, _, reader, controller, queue, _ = asking
card = await start(asking)
await controller.answer_question(card["task_id"])
await receive(queue, "result") # bounded to 3 seconds, not the 60-second timeout
with pytest.raises(InputError, match="does not exist"):
await controller.answer_question(card["task_id"], "late conflicting answer")
with pytest.raises(InputError, match="does not exist"):
await controller.answer_question(card["task_id"])
history = (await reader.get(controller.session_id)).messages
assert tool_results(history) == [{"task_id": card["task_id"], "status": "skipped"}]
assert late_answers(history) == []
assert controller.subscribe()[1]["questions"] == []
assert controller.questions == {}
@pytest.mark.asyncio
async def test_skip_after_timeout_delivers_one_result(asking):
_, _, reader, controller, queue, _ = asking
card = await start(asking)
await receive(queue, "result")
if controller.active_task:
await controller.active_task
item = controller.questions[card["task_id"]]
background_task = item.background_task
await controller.answer_question(card["task_id"])
with pytest.raises(InputError, match="does not exist"):
await controller.answer_question(card["task_id"], "late conflicting answer")
await background_task
await receive(queue, "question_closed")
await receive(queue, "result")
history = (await reader.get(controller.session_id)).messages
assert tool_results(history) == [{"status": "pending", "task_id": "ask"}]
assert background_results(history) == [{"task_id": "ask", "status": "skipped"}]
assert controller.subscribe()[1]["questions"] == []
assert item.future is None and item.background_task is None
@pytest.mark.asyncio
@pytest.mark.parametrize("skip_first", [False, True])
async def test_answer_and_skip_first_submission_wins(asking, skip_first):
_, _, reader, controller, queue, _ = asking
card = await start(asking)
answers = [None, "MySQL"] if skip_first else ["MySQL", None]
results = await asyncio.gather(
*(controller.answer_question(card["task_id"], a) for a in answers),
return_exceptions=True,
)
assert results[0] is None
assert isinstance(results[1], InputError)
assert controller.questions == {}
await receive(queue, "result")
history = (await reader.get(controller.session_id)).messages
assert len(tool_results(history)) == 1
assert tool_results(history)[0]["status"] == ("skipped" if skip_first else "answered")
assert late_answers(history) == []
def test_websocket_skip_errors_and_reconnect(tmp_path, monkeypatch):
monkeypatch.setattr(questions, "ASK_USER_WAIT_SECONDS", 60)
service, _, reader = make_service(tmp_path)
with TestClient(create_app(runtime_service=service, session_reader=reader)) as client:
with client.websocket_connect("/ws/v1/chat/skip-test") as socket:
socket.receive_json()
socket.send_json({"type": "question_skip", "task_id": "unknown"})
assert socket.receive_json()["task_id"] == "unknown"
socket.send_json({"type": "message", "content": "Ask", "config": {
"runtime": {"workspace": str(tmp_path)},
}})
frame = socket.receive_json()
while frame["type"] != "question_request":
frame = socket.receive_json()
socket.send_json({"type": "question_skip", "task_id": frame["task_id"]})
frame = socket.receive_json()
while frame["type"] != "question_closed":
frame = socket.receive_json()
while socket.receive_json()["type"] != "result":
pass
with client.websocket_connect("/ws/v1/chat/skip-test") as socket:
assert socket.receive_json()["questions"] == []
@pytest.mark.asyncio
@pytest.mark.parametrize("running", [False, True])
async def test_background_deadline_delivers_skip_and_closes_card(asking, monkeypatch, running):
monkeypatch.setattr(questions, "ASK_USER_BACKGROUND_WAIT_SECONDS", 0.1)
_, provider, reader, controller, queue, _ = asking
if running:
provider.release.clear()
card = await start(asking)
await asyncio.wait_for(provider.after_tool.wait(), 3)
item = controller.questions[card["task_id"]]
background_task = item.background_task
assert background_task is not None and item.future is not None
assert not item.future.done()
assert controller.subscribe()[1]["questions"] # pending is not skipped
await asyncio.wait_for(background_task, 3)
assert await receive(queue, "question_closed") == {
"type": "question_closed", "session_id": controller.session_id, "task_id": "ask",
}
assert not controller.questions
provider.release.set()
async with asyncio.timeout(3):
while controller.active_task:
await asyncio.sleep(0.001)
history = (await reader.get(controller.session_id)).messages
assert tool_results(history) == [{"status": "pending", "task_id": "ask"}]
assert background_results(history) == [{"task_id": "ask", "status": "skipped"}]
assert item.future is None and item.background_task is None
with pytest.raises(InputError, match="does not exist"):
await controller.answer_question("ask", "too late")
@pytest.mark.asyncio
async def test_background_expiration_after_stop_is_queued_until_resume(asking, monkeypatch):
monkeypatch.setattr(questions, "ASK_USER_BACKGROUND_WAIT_SECONDS", 0.1)
_, provider, reader, controller, queue, _ = asking
provider.release.clear()
card = await start(asking)
await asyncio.wait_for(provider.after_tool.wait(), 3)
background_task = controller.questions[card["task_id"]].background_task
await controller.cancel()
await asyncio.wait_for(background_task, 3)
assert controller.active_task is None and controller._paused
assert not controller.questions
assert len(controller.pending) == 1
assert controller.pending[0].request.message_id == "ask_user:ask"
assert '"status": "skipped"' in controller.pending[0].request.content
provider.release.set()
await controller.steer(controller.pending[0].request.message_id)
await receive(queue, "result")
assert background_results((await reader.get(controller.session_id)).messages) == [
{"task_id": "ask", "status": "skipped"},
]
@pytest.mark.asyncio
async def test_disconnect_does_not_reset_background_deadline(asking, monkeypatch):
monkeypatch.setattr(questions, "ASK_USER_BACKGROUND_WAIT_SECONDS", 0.1)
_, provider, _, controller, queue, _ = asking
card = await start(asking)
await asyncio.wait_for(provider.after_tool.wait(), 3)
background_task = controller.questions[card["task_id"]].background_task
controller.unsubscribe(queue)
new_queue, snapshot = controller.subscribe()
assert snapshot["questions"][0]["task_id"] == card["task_id"]
controller.unsubscribe(new_queue)
await asyncio.wait_for(background_task, 3)
assert controller.subscribe()[1]["questions"] == []
@pytest.mark.asyncio
async def test_close_cancels_background_wait_without_delivering_skip(asking, monkeypatch):
monkeypatch.setattr(questions, "ASK_USER_BACKGROUND_WAIT_SECONDS", 0.1)
_, provider, reader, controller, queue, _ = asking
card = await start(asking)
await asyncio.wait_for(provider.after_tool.wait(), 3)
item = controller.questions[card["task_id"]]
background_task = item.background_task
future = item.future
await controller.close()
assert background_task.cancelled()
assert future.cancelled()
assert not controller.questions and not controller.pending
assert background_results((await reader.get(controller.session_id)).messages) == []
@pytest.mark.asyncio
@pytest.mark.parametrize("answer_first", [True, False])
async def test_background_answer_and_deadline_deliver_only_one_result(asking, monkeypatch, answer_first):
monkeypatch.setattr(questions, "ASK_USER_BACKGROUND_WAIT_SECONDS", 0)
_, provider, reader, controller, queue, _ = asking
card = await start(asking)
item = controller.questions[card["task_id"]]
# Hold the completion callback at the deadline; submission must not overwrite
# a Future that has already completed, even while its card still exists.
entered = asyncio.Event()
release = asyncio.Event()
original_finish = controller._finish_question
async def finish(question):
entered.set()
await release.wait()
await original_finish(question)
# wait() has captured the original callback, so exercise the same waiter
# directly with a controlled boundary and the real submission handler.
await controller.cancel()
item.future = asyncio.get_running_loop().create_future()
item.background_task = asyncio.create_task(item._wait_background(finish, asyncio.get_running_loop().time()))
background_task = item.background_task
if answer_first:
await controller.answer_question(card["task_id"], "boundary answer")
await asyncio.wait_for(entered.wait(), 3)
with pytest.raises(InputError, match="does not exist"):
await controller.answer_question(card["task_id"], "duplicate")
release.set()
await asyncio.wait_for(background_task, 3)
assert len(controller.pending) == 1 # paused by Stop, no automatic restart
result = json.loads(controller.pending[0].request.content.split("\n\n", 1)[1])
assert result == ({"task_id": "ask", "status": "answered", "answer": "boundary answer"}
if answer_first else {"task_id": "ask", "status": "skipped"})
assert not controller.questions
@pytest.mark.parametrize("action", ["answer", "skip", "expire"])
def test_websocket_background_completion_and_reconnect(tmp_path, monkeypatch, action):
monkeypatch.setattr(questions, "ASK_USER_WAIT_SECONDS", 0.02)
monkeypatch.setattr(questions, "ASK_USER_BACKGROUND_WAIT_SECONDS", 0.15)
service, provider, reader = make_service(tmp_path)
app = create_app(runtime_service=service, session_reader=reader)
with TestClient(app) as client:
session_id = client.portal.call(reader.create, tmp_path)
with client.websocket_connect(f"/ws/v1/chat/{session_id}") as socket:
socket.receive_json()
socket.send_json({"type": "message", "content": "Ask", "config": {
"runtime": {"workspace": str(tmp_path)},
}})
while socket.receive_json()["type"] != "question_request":
pass
while socket.receive_json()["type"] != "result":
pass
if action != "expire":
socket.send_json({
"type": "question_answer" if action == "answer" else "question_skip",
"task_id": "ask",
**({"answer": "MySQL"} if action == "answer" else {}),
})
closed = socket.receive_json()
while closed["type"] != "question_closed":
closed = socket.receive_json()
assert closed["task_id"] == "ask"
while socket.receive_json()["type"] != "result":
pass
with client.websocket_connect(f"/ws/v1/chat/{session_id}") as socket:
assert socket.receive_json()["questions"] == []
history = client.portal.call(reader.get, session_id).messages
assert background_results(history) == [
{"task_id": "ask", "status": "answered", "answer": "MySQL"}
if action == "answer" else {"task_id": "ask", "status": "skipped"},
]
@pytest.mark.asyncio
async def test_answer_and_expiration_in_different_sessions_are_isolated(asking, monkeypatch):
monkeypatch.setattr(questions, "ASK_USER_BACKGROUND_WAIT_SECONDS", 0.15)
service, provider, reader, first, first_queue, workspace = asking
second = service.get_session_controller(await reader.create(workspace))
second_queue, _ = second.subscribe()
await asyncio.gather(start(asking), start((service, provider, reader, second, second_queue, workspace)))
await asyncio.gather(receive(first_queue, "result"), receive(second_queue, "result"))
first_wait = first.questions["ask"].background_task
second_wait = second.questions["ask"].background_task
await first.answer_question("ask", "only first session")
await asyncio.wait_for(asyncio.gather(first_wait, second_wait), 3)
await asyncio.gather(receive(first_queue, "result"), receive(second_queue, "result"))
assert background_results((await reader.get(first.session_id)).messages) == [
{"task_id": "ask", "status": "answered", "answer": "only first session"},
]
assert background_results((await reader.get(second.session_id)).messages) == [
{"task_id": "ask", "status": "skipped"},
]
assert not first.questions and not second.questions