| import pytest |
|
|
| from arena.event_bus import EventBus |
| from arena.run_journal import InvalidRunStatusTransition, RunJournal |
| from arena.run_models import RunEvent, RunRequest, RunTask |
| from arena.run_store import InMemoryRunStore |
|
|
|
|
| def test_run_journal_owns_status_transitions_and_streams() -> None: |
| journal = RunJournal(run_store=InMemoryRunStore(), event_bus=EventBus()) |
| access = journal.create_run(RunRequest(prompt="Build a CLI")) |
| queue = journal.subscribe_events(access.run.id) |
|
|
| journal.transition(access.run.id, "planning") |
| journal.publish(access.run.id, {"kind": "run_started", "run_id": access.run.id}) |
|
|
| view = journal.get_run(access.run.id) |
|
|
| assert view.status == "planning" |
| assert queue.get(timeout=1)["kind"] == "run_started" |
|
|
|
|
| def test_run_journal_rejects_invalid_status_transition() -> None: |
| journal = RunJournal(run_store=InMemoryRunStore(), event_bus=EventBus()) |
| access = journal.create_run(RunRequest(prompt="Build a CLI")) |
|
|
| with pytest.raises(InvalidRunStatusTransition): |
| journal.transition(access.run.id, "completed") |
|
|
|
|
| def test_run_journal_persists_runtime_task_and_event_updates() -> None: |
| journal = RunJournal(run_store=InMemoryRunStore(), event_bus=EventBus()) |
| access = journal.create_run(RunRequest(prompt="Build a CLI")) |
| task = RunTask(id="task-1", title="Build codebase", status="active") |
| event = RunEvent(id="event-1", message="Coordinator started work") |
|
|
| journal.publish(access.run.id, {"kind": "task_created", "task": task.model_dump()}) |
| journal.publish(access.run.id, {"kind": "event", "event": event.model_dump(mode="json")}) |
|
|
| view = journal.get_run(access.run.id) |
|
|
| assert view.tasks == [task] |
| assert [item.id for item in view.events] == ["event-1"] |
|
|
|
|
| def test_run_journal_cancels_active_tasks_and_closes_stream() -> None: |
| journal = RunJournal(run_store=InMemoryRunStore(), event_bus=EventBus()) |
| access = journal.create_run(RunRequest(prompt="Build a CLI")) |
| queue = journal.subscribe_events(access.run.id) |
| journal.transition(access.run.id, "working") |
| journal.publish( |
| access.run.id, |
| { |
| "kind": "task_created", |
| "task": RunTask(id="task-1", title="Build codebase", status="active").model_dump(), |
| }, |
| ) |
|
|
| view = journal.cancel_run(access.run.id) |
|
|
| assert view.status == "interrupted" |
| assert view.tasks[0].status == "interrupted" |
| assert queue.get(timeout=1)["kind"] == "task_created" |
| assert queue.get(timeout=1) is None |
|
|
