izackp/takopi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: introduce runner protocol and normalized event model (#7)

Browse Source
This commit is contained in:
banteg
2026-01-01 01:13:55 +04:00
committed by GitHub
parent a9f8967bf4
commit d296c0dbf1
36 changed files with 4749 additions and 1836 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Test helpers package."""
tests/factories.py
+64
View File
@@ -0,0 +1,64 @@
from __future__ import annotations
from typing import Any
from takopi.model import (
Action,
ActionEvent,
ActionKind,
EngineId,
ResumeToken,
StartedEvent,
TakopiEvent,
)
def session_started(engine: str, value: str, title: str = "Codex") -> TakopiEvent:
engine_id = EngineId(engine)
return StartedEvent(
engine=engine_id,
resume=ResumeToken(engine=engine_id, value=value),
title=title,
)
def action_started(
action_id: str,
kind: ActionKind,
title: str,
detail: dict[str, Any] | None = None,
engine: str = "codex",
) -> TakopiEvent:
engine_id = EngineId(engine)
return ActionEvent(
engine=engine_id,
action=Action(
id=action_id,
kind=kind,
title=title,
detail=detail or {},
),
phase="started",
)
def action_completed(
action_id: str,
kind: ActionKind,
title: str,
ok: bool,
detail: dict[str, Any] | None = None,
engine: str = "codex",
) -> TakopiEvent:
engine_id = EngineId(engine)
return ActionEvent(
engine=engine_id,
action=Action(
id=action_id,
kind=kind,
title=title,
detail=detail or {},
),
phase="completed",
ok=ok,
)
tests/test_exec_bridge.py
+430 -267
View File
@@ -1,106 +1,116 @@
import uuid
import anyio
import pytest
from takopi.exec_bridge import (
extract_session_id,
prepare_telegram,
resolve_resume_session,
truncate_for_telegram,
)
from takopi import engines
from takopi.markdown import prepare_telegram, truncate_for_telegram
from takopi.model import EngineId, ResumeToken, TakopiEvent
from takopi.runners.codex import CodexRunner
from takopi.runners.mock import Advance, Emit, Raise, Return, ScriptRunner, Sleep, Wait
from tests.factories import action_completed, action_started
CODEX_ENGINE = EngineId("codex")
def _patch_config(monkeypatch, config):
from pathlib import Path
from takopi import exec_bridge
from takopi import cli
monkeypatch.setattr(
exec_bridge,
cli,
"load_telegram_config",
lambda: (config, Path("takopi.toml")),
)
def test_parse_bridge_config_rejects_empty_token(monkeypatch) -> None:
from takopi import exec_bridge
from takopi import cli
_patch_config(monkeypatch, {"bot_token": " ", "chat_id": 123})
with pytest.raises(exec_bridge.ConfigError, match="bot_token"):
exec_bridge._parse_bridge_config(final_notify=True, profile=None)
with pytest.raises(cli.ConfigError, match="bot_token"):
cli._parse_bridge_config(
final_notify=True,
backend=engines.get_backend("codex"),
engine_overrides={},
)
def test_parse_bridge_config_rejects_string_chat_id(monkeypatch) -> None:
from takopi import exec_bridge
from takopi import cli
_patch_config(monkeypatch, {"bot_token": "token", "chat_id": "123"})
with pytest.raises(exec_bridge.ConfigError, match="chat_id"):
exec_bridge._parse_bridge_config(final_notify=True, profile=None)
with pytest.raises(cli.ConfigError, match="chat_id"):
cli._parse_bridge_config(
final_notify=True,
backend=engines.get_backend("codex"),
engine_overrides={},
)
def test_extract_session_id_finds_uuid_v7() -> None:
def test_codex_extract_resume_finds_command() -> None:
uuid = "019b66fc-64c2-7a71-81cd-081c504cfeb2"
text = f"resume: `{uuid}`"
runner = CodexRunner(codex_cmd="codex", extra_args=[])
text = f"`codex resume {uuid}`"
assert extract_session_id(text) == uuid
assert runner.extract_resume(text) == ResumeToken(engine=CODEX_ENGINE, value=uuid)
def test_extract_session_id_requires_resume_line() -> None:
uuid = "019b66fc-64c2-7a71-81cd-081c504cfeb2"
text = f"here is a uuid {uuid}"
assert extract_session_id(text) is None
def test_extract_session_id_uses_last_resume_line() -> None:
def test_codex_extract_resume_uses_last_resume_line() -> None:
uuid_first = "019b66fc-64c2-7a71-81cd-081c504cfeb2"
uuid_last = "123e4567-e89b-12d3-a456-426614174000"
text = f"resume: `{uuid_first}`\n\nresume: `{uuid_last}`"
runner = CodexRunner(codex_cmd="codex", extra_args=[])
text = f"`codex resume {uuid_first}`\n\n`codex resume {uuid_last}`"
assert extract_session_id(text) == uuid_last
def test_extract_session_id_ignores_malformed_resume_line() -> None:
text = "resume: not-a-uuid"
assert extract_session_id(text) is None
def test_resolve_resume_session_prefers_message_text() -> None:
uuid_message = "123e4567-e89b-12d3-a456-426614174000"
uuid_reply = "019b66fc-64c2-7a71-81cd-081c504cfeb2"
assert (
resolve_resume_session(f"resume: `{uuid_message}`", f"resume: `{uuid_reply}`")
== uuid_message
assert runner.extract_resume(text) == ResumeToken(
engine=CODEX_ENGINE, value=uuid_last
)
def test_resolve_resume_session_uses_reply_when_missing() -> None:
uuid_reply = "019b66fc-64c2-7a71-81cd-081c504cfeb2"
def test_codex_extract_resume_ignores_malformed_resume_line() -> None:
runner = CodexRunner(codex_cmd="codex", extra_args=[])
text = "codex resume"
assert (
resolve_resume_session("no resume here", f"resume: `{uuid_reply}`")
== uuid_reply
)
assert runner.extract_resume(text) is None
def test_codex_extract_resume_accepts_plain_line() -> None:
uuid = "019b66fc-64c2-7a71-81cd-081c504cfeb2"
runner = CodexRunner(codex_cmd="codex", extra_args=[])
text = f"codex resume {uuid}"
assert runner.extract_resume(text) == ResumeToken(engine=CODEX_ENGINE, value=uuid)
def test_codex_extract_resume_accepts_uuid7() -> None:
uuid7 = getattr(uuid, "uuid7", None)
assert uuid7 is not None
token = str(uuid7())
runner = CodexRunner(codex_cmd="codex", extra_args=[])
text = f"`codex resume {token}`"
assert runner.extract_resume(text) == ResumeToken(engine=CODEX_ENGINE, value=token)
def test_truncate_for_telegram_preserves_resume_line() -> None:
uuid = "019b66fc-64c2-7a71-81cd-081c504cfeb2"
md = ("x" * 10_000) + f"\nresume: `{uuid}`"
md = ("x" * 10_000) + f"\n`codex resume {uuid}`"
out = truncate_for_telegram(md, 400)
runner = CodexRunner(codex_cmd="codex", extra_args=[])
out = truncate_for_telegram(md, 400, is_resume_line=runner.is_resume_line)
assert len(out) <= 400
assert uuid in out
assert out.rstrip().endswith(f"resume: `{uuid}`")
assert f"codex resume {uuid}" in out
assert out.rstrip().endswith(f"`codex resume {uuid}`")
def test_truncate_for_telegram_keeps_last_non_empty_line() -> None:
md = "intro\n\n" + ("x" * 500) + "\nlast line"
out = truncate_for_telegram(md, 120)
out = truncate_for_telegram(md, 120, is_resume_line=lambda _line: False)
assert len(out) <= 120
assert out.rstrip().endswith("last line")
@@ -168,18 +178,19 @@ class _FakeBot:
self.delete_calls.append({"chat_id": chat_id, "message_id": message_id})
return True
async def get_updates(
self,
offset: int | None,
timeout_s: int = 50,
allowed_updates: list[str] | None = None,
) -> list[dict] | None:
_ = offset
_ = timeout_s
_ = allowed_updates
return []
class _FakeRunner:
def __init__(self, *, answer: str, saw_agent_message: bool = True) -> None:
self._answer = answer
self._saw_agent_message = saw_agent_message
async def run_serialized(self, *_args, **_kwargs) -> tuple[str, str, bool]:
return (
"019b66fc-64c2-7a71-81cd-081c504cfeb2",
self._answer,
self._saw_agent_message,
)
async def close(self) -> None:
return None
class _FakeClock:
@@ -211,54 +222,28 @@ class _FakeClock:
await self._sleep_event.wait()
class _FakeRunnerWithEvents:
def __init__(
self,
*,
events: list[dict],
times: list[float],
clock: _FakeClock,
answer: str = "ok",
session_id: str = "019b66fc-64c2-7a71-81cd-081c504cfeb2",
advance_after: float | None = None,
hold: anyio.Event | None = None,
) -> None:
self._events = events
self._times = times
self._clock = clock
self._answer = answer
self._session_id = session_id
self._advance_after = advance_after
self._hold = hold
async def run_serialized(self, *_args, **kwargs) -> tuple[str, str, bool]:
on_event = kwargs.get("on_event")
if on_event is not None:
for when, event in zip(self._times, self._events, strict=False):
self._clock.set(when)
await on_event(event)
await anyio.sleep(0)
if self._advance_after is not None:
self._clock.set(self._advance_after)
await anyio.sleep(0)
if self._hold is not None:
await self._hold.wait()
return (self._session_id, self._answer, True)
def _return_runner(
*, answer: str = "ok", resume_value: str | None = None
) -> ScriptRunner:
return ScriptRunner(
[Return(answer=answer)],
engine=CODEX_ENGINE,
resume_value=resume_value,
)
@pytest.mark.anyio
async def test_final_notify_sends_loud_final_message() -> None:
from takopi.exec_bridge import BridgeConfig, handle_message
from takopi.bridge import BridgeConfig, handle_message
bot = _FakeBot()
runner = _FakeRunner(answer="ok")
runner = _return_runner(answer="ok")
cfg = BridgeConfig(
bot=bot, # type: ignore[arg-type]
runner=runner, # type: ignore[arg-type]
bot=bot,
runner=runner,
chat_id=123,
final_notify=True,
startup_msg="",
max_concurrency=1,
)
await handle_message(
@@ -266,7 +251,7 @@ async def test_final_notify_sends_loud_final_message() -> None:
chat_id=123,
user_msg_id=10,
text="hi",
resume_session=None,
resume_token=None,
)
assert len(bot.send_calls) == 2
@@ -275,18 +260,47 @@ async def test_final_notify_sends_loud_final_message() -> None:
@pytest.mark.anyio
async def test_new_final_message_forces_notification_when_too_long_to_edit() -> None:
from takopi.exec_bridge import BridgeConfig, handle_message
async def test_handle_message_strips_resume_line_from_prompt() -> None:
from takopi.bridge import BridgeConfig, handle_message
bot = _FakeBot()
runner = _FakeRunner(answer="x" * 10_000)
runner = ScriptRunner([Return(answer="ok")], engine=CODEX_ENGINE)
cfg = BridgeConfig(
bot=bot, # type: ignore[arg-type]
runner=runner, # type: ignore[arg-type]
bot=bot,
runner=runner,
chat_id=123,
final_notify=True,
startup_msg="",
)
resume = ResumeToken(engine=CODEX_ENGINE, value="sid")
text = "do this\n`codex resume sid`\nand that"
await handle_message(
cfg,
chat_id=123,
user_msg_id=10,
text=text,
resume_token=resume,
)
assert runner.calls
prompt, passed_resume = runner.calls[0]
assert prompt == "do this\nand that"
assert passed_resume == resume
@pytest.mark.anyio
async def test_new_final_message_forces_notification_when_too_long_to_edit() -> None:
from takopi.bridge import BridgeConfig, handle_message
bot = _FakeBot()
runner = _return_runner(answer="x" * 10_000)
cfg = BridgeConfig(
bot=bot,
runner=runner,
chat_id=123,
final_notify=False,
startup_msg="",
max_concurrency=1,
)
await handle_message(
@@ -294,7 +308,7 @@ async def test_new_final_message_forces_notification_when_too_long_to_edit() ->
chat_id=123,
user_msg_id=10,
text="hi",
resume_session=None,
resume_token=None,
)
assert len(bot.send_calls) == 2
@@ -304,43 +318,30 @@ async def test_new_final_message_forces_notification_when_too_long_to_edit() ->
@pytest.mark.anyio
async def test_progress_edits_are_rate_limited() -> None:
from takopi.exec_bridge import BridgeConfig, handle_message
from takopi.bridge import BridgeConfig, handle_message
bot = _FakeBot()
clock = _FakeClock()
events = [
{
"type": "item.started",
"item": {
"id": "item_0",
"type": "command_execution",
"command": "echo 1",
"status": "in_progress",
},
},
{
"type": "item.started",
"item": {
"id": "item_1",
"type": "command_execution",
"command": "echo 2",
"status": "in_progress",
},
},
events: list[TakopiEvent] = [
action_started("item_0", "command", "echo 1"),
action_started("item_1", "command", "echo 2"),
]
runner = _FakeRunnerWithEvents(
events=events,
times=[0.2, 0.4],
clock=clock,
advance_after=1.0,
runner = ScriptRunner(
[
Emit(events[0], at=0.2),
Emit(events[1], at=0.4),
Advance(1.0),
Return(answer="ok"),
],
engine=CODEX_ENGINE,
advance=clock.set,
)
cfg = BridgeConfig(
bot=bot, # type: ignore[arg-type]
runner=runner, # type: ignore[arg-type]
bot=bot,
runner=runner,
chat_id=123,
final_notify=True,
startup_msg="",
max_concurrency=1,
)
await handle_message(
@@ -348,7 +349,7 @@ async def test_progress_edits_are_rate_limited() -> None:
chat_id=123,
user_msg_id=10,
text="hi",
resume_session=None,
resume_token=None,
clock=clock,
sleep=clock.sleep,
progress_edit_every=1.0,
@@ -360,45 +361,31 @@ async def test_progress_edits_are_rate_limited() -> None:
@pytest.mark.anyio
async def test_progress_edits_do_not_sleep_again_without_new_events() -> None:
from takopi.exec_bridge import BridgeConfig, handle_message
from takopi.bridge import BridgeConfig, handle_message
bot = _FakeBot()
clock = _FakeClock()
hold = anyio.Event()
events = [
{
"type": "item.started",
"item": {
"id": "item_0",
"type": "command_execution",
"command": "echo 1",
"status": "in_progress",
},
},
{
"type": "item.started",
"item": {
"id": "item_1",
"type": "command_execution",
"command": "echo 2",
"status": "in_progress",
},
},
events: list[TakopiEvent] = [
action_started("item_0", "command", "echo 1"),
action_started("item_1", "command", "echo 2"),
]
runner = _FakeRunnerWithEvents(
events=events,
times=[0.2, 0.4],
clock=clock,
advance_after=None,
hold=hold,
runner = ScriptRunner(
[
Emit(events[0], at=0.2),
Emit(events[1], at=0.4),
Wait(hold),
Return(answer="ok"),
],
engine=CODEX_ENGINE,
advance=clock.set,
)
cfg = BridgeConfig(
bot=bot, # type: ignore[arg-type]
runner=runner, # type: ignore[arg-type]
bot=bot,
runner=runner,
chat_id=123,
final_notify=True,
startup_msg="",
max_concurrency=1,
)
async def run_handle_message() -> None:
@@ -407,7 +394,7 @@ async def test_progress_edits_do_not_sleep_again_without_new_events() -> None:
chat_id=123,
user_msg_id=10,
text="hi",
resume_session=None,
resume_token=None,
clock=clock,
sleep=clock.sleep,
progress_edit_every=1.0,
@@ -443,46 +430,37 @@ async def test_progress_edits_do_not_sleep_again_without_new_events() -> None:
@pytest.mark.anyio
async def test_bridge_flow_sends_progress_edits_and_final_resume() -> None:
from takopi.exec_bridge import BridgeConfig, handle_message
from takopi.bridge import BridgeConfig, handle_message
bot = _FakeBot()
clock = _FakeClock()
events = [
{
"type": "item.started",
"item": {
"id": "item_0",
"type": "command_execution",
"command": "echo ok",
"status": "in_progress",
},
},
{
"type": "item.completed",
"item": {
"id": "item_0",
"type": "command_execution",
"command": "echo ok",
"exit_code": 0,
"status": "completed",
},
},
events: list[TakopiEvent] = [
action_started("item_0", "command", "echo ok"),
action_completed(
"item_0",
"command",
"echo ok",
ok=True,
detail={"exit_code": 0},
),
]
session_id = "123e4567-e89b-12d3-a456-426614174000"
runner = _FakeRunnerWithEvents(
events=events,
times=[0.0, 2.1],
clock=clock,
answer="done",
session_id=session_id,
runner = ScriptRunner(
[
Emit(events[0], at=0.0),
Emit(events[1], at=2.1),
Return(answer="done"),
],
engine=CODEX_ENGINE,
advance=clock.set,
resume_value=session_id,
)
cfg = BridgeConfig(
bot=bot, # type: ignore[arg-type]
runner=runner, # type: ignore[arg-type]
bot=bot,
runner=runner,
chat_id=123,
final_notify=True,
startup_msg="",
max_concurrency=1,
)
await handle_message(
@@ -490,7 +468,7 @@ async def test_bridge_flow_sends_progress_edits_and_final_resume() -> None:
chat_id=123,
user_msg_id=42,
text="do it",
resume_session=None,
resume_token=None,
clock=clock,
sleep=clock.sleep,
progress_edit_every=1.0,
@@ -500,23 +478,22 @@ async def test_bridge_flow_sends_progress_edits_and_final_resume() -> None:
assert "working" in bot.send_calls[0]["text"]
assert len(bot.edit_calls) >= 1
assert session_id in bot.send_calls[-1]["text"]
assert "resume:" in bot.send_calls[-1]["text"].lower()
assert "codex resume" in bot.send_calls[-1]["text"].lower()
assert len(bot.delete_calls) == 1
@pytest.mark.anyio
async def test_handle_cancel_without_reply_prompts_user() -> None:
from takopi.exec_bridge import BridgeConfig, _handle_cancel
from takopi.bridge import BridgeConfig, _handle_cancel
bot = _FakeBot()
runner = _FakeRunner(answer="ok")
runner = _return_runner(answer="ok")
cfg = BridgeConfig(
bot=bot, # type: ignore[arg-type]
runner=runner, # type: ignore[arg-type]
bot=bot,
runner=runner,
chat_id=123,
final_notify=True,
startup_msg="",
max_concurrency=1,
)
msg = {"chat": {"id": 123}, "message_id": 10}
running_tasks: dict = {}
@@ -529,17 +506,16 @@ async def test_handle_cancel_without_reply_prompts_user() -> None:
@pytest.mark.anyio
async def test_handle_cancel_with_no_progress_message_says_nothing_running() -> None:
from takopi.exec_bridge import BridgeConfig, _handle_cancel
from takopi.bridge import BridgeConfig, _handle_cancel
bot = _FakeBot()
runner = _FakeRunner(answer="ok")
runner = _return_runner(answer="ok")
cfg = BridgeConfig(
bot=bot, # type: ignore[arg-type]
runner=runner, # type: ignore[arg-type]
bot=bot,
runner=runner,
chat_id=123,
final_notify=True,
startup_msg="",
max_concurrency=1,
)
msg = {
"chat": {"id": 123},
@@ -556,17 +532,16 @@ async def test_handle_cancel_with_no_progress_message_says_nothing_running() ->
@pytest.mark.anyio
async def test_handle_cancel_with_finished_task_says_nothing_running() -> None:
from takopi.exec_bridge import BridgeConfig, _handle_cancel
from takopi.bridge import BridgeConfig, _handle_cancel
bot = _FakeBot()
runner = _FakeRunner(answer="ok")
runner = _return_runner(answer="ok")
cfg = BridgeConfig(
bot=bot, # type: ignore[arg-type]
runner=runner, # type: ignore[arg-type]
bot=bot,
runner=runner,
chat_id=123,
final_notify=True,
startup_msg="",
max_concurrency=1,
)
progress_id = 99
msg = {
@@ -584,17 +559,16 @@ async def test_handle_cancel_with_finished_task_says_nothing_running() -> None:
@pytest.mark.anyio
async def test_handle_cancel_cancels_running_task() -> None:
from takopi.exec_bridge import BridgeConfig, _handle_cancel
from takopi.bridge import BridgeConfig, _handle_cancel
bot = _FakeBot()
runner = _FakeRunner(answer="ok")
runner = _return_runner(answer="ok")
cfg = BridgeConfig(
bot=bot, # type: ignore[arg-type]
runner=runner, # type: ignore[arg-type]
bot=bot,
runner=runner,
chat_id=123,
final_notify=True,
startup_msg="",
max_concurrency=1,
)
progress_id = 42
msg = {
@@ -603,49 +577,33 @@ async def test_handle_cancel_cancels_running_task() -> None:
"reply_to_message": {"message_id": progress_id},
}
from takopi.exec_bridge import RunningTask
from takopi.bridge import RunningTask
cancelled_event = anyio.Event()
cancel_scope = anyio.CancelScope()
running_task = RunningTask(scope=cancel_scope)
async def sleeper() -> None:
with cancel_scope:
try:
await anyio.sleep(10)
except anyio.get_cancelled_exc_class():
cancelled_event.set()
return
async with anyio.create_task_group() as tg:
tg.start_soon(sleeper)
running_tasks = {progress_id: running_task}
await _handle_cancel(cfg, msg, running_tasks)
await cancelled_event.wait()
running_task = RunningTask()
running_tasks = {progress_id: running_task}
await _handle_cancel(cfg, msg, running_tasks)
assert running_task.cancel_requested.is_set() is True
assert len(bot.send_calls) == 0 # No error message sent
@pytest.mark.anyio
async def test_handle_cancel_only_cancels_matching_progress_message() -> None:
from takopi.exec_bridge import BridgeConfig, _handle_cancel
from takopi.bridge import BridgeConfig, _handle_cancel
bot = _FakeBot()
runner = _FakeRunner(answer="ok")
runner = _return_runner(answer="ok")
cfg = BridgeConfig(
bot=bot, # type: ignore[arg-type]
runner=runner, # type: ignore[arg-type]
bot=bot,
runner=runner,
chat_id=123,
final_notify=True,
startup_msg="",
max_concurrency=1,
)
from takopi.exec_bridge import RunningTask
from takopi.bridge import RunningTask
scope_first = anyio.CancelScope()
scope_second = anyio.CancelScope()
task_first = RunningTask(scope=scope_first)
task_second = RunningTask(scope=scope_second)
task_first = RunningTask()
task_second = RunningTask()
msg = {
"chat": {"id": 123},
"message_id": 10,
@@ -655,37 +613,56 @@ async def test_handle_cancel_only_cancels_matching_progress_message() -> None:
await _handle_cancel(cfg, msg, running_tasks)
assert scope_first.cancel_called is True
assert scope_second.cancel_called is False
assert task_first.cancel_requested.is_set() is True
assert task_second.cancel_requested.is_set() is False
assert len(bot.send_calls) == 0
class _FakeRunnerCancellable:
def __init__(self, session_id: str = "019b66fc-64c2-7a71-81cd-081c504cfeb2"):
self._session_id = session_id
def test_cancel_command_accepts_extra_text() -> None:
from takopi.bridge import _is_cancel_command
async def run_serialized(self, *_args, **kwargs) -> tuple[str, str, bool]:
on_event = kwargs.get("on_event")
if on_event:
await on_event({"type": "thread.started", "thread_id": self._session_id})
await anyio.sleep(10) # Will be cancelled
return (self._session_id, "ok", True)
assert _is_cancel_command("/cancel now") is True
assert _is_cancel_command("/cancel@takopi please") is True
assert _is_cancel_command("/cancelled") is False
def test_resume_attempt_does_not_trigger_on_plain_resume_word() -> None:
from takopi.bridge import _resume_attempt
attempt, engine = _resume_attempt("resume abc123")
assert attempt is False
assert engine is None
def test_resume_warning_for_other_engine() -> None:
from takopi.bridge import _resume_attempt, _resume_warning_text
attempt, engine = _resume_attempt("claude resume abc123")
assert attempt is True
assert engine == "claude"
warning = _resume_warning_text(engine, "codex")
assert "claude" in warning.lower()
assert "codex" in warning.lower()
@pytest.mark.anyio
async def test_handle_message_cancelled_renders_cancelled_state() -> None:
from takopi.exec_bridge import BridgeConfig, handle_message
from takopi.bridge import BridgeConfig, handle_message
bot = _FakeBot()
session_id = "019b66fc-64c2-7a71-81cd-081c504cfeb2"
runner = _FakeRunnerCancellable(session_id=session_id)
hold = anyio.Event()
runner = ScriptRunner(
[Wait(hold)],
engine=CODEX_ENGINE,
resume_value=session_id,
)
cfg = BridgeConfig(
bot=bot, # type: ignore[arg-type]
runner=runner, # type: ignore[arg-type]
bot=bot,
runner=runner,
chat_id=123,
final_notify=True,
startup_msg="",
max_concurrency=1,
)
running_tasks: dict = {}
@@ -695,7 +672,7 @@ async def test_handle_message_cancelled_renders_cancelled_state() -> None:
chat_id=123,
user_msg_id=10,
text="do something",
resume_session=None,
resume_token=None,
running_tasks=running_tasks,
)
@@ -706,10 +683,196 @@ async def test_handle_message_cancelled_renders_cancelled_state() -> None:
break
await anyio.sleep(0)
assert running_tasks
running_tasks[next(iter(running_tasks))].scope.cancel()
running_task = running_tasks[next(iter(running_tasks))]
with anyio.fail_after(1):
await running_task.resume_ready.wait()
running_task.cancel_requested.set()
assert len(bot.send_calls) == 1 # Progress message
assert len(bot.edit_calls) >= 1
last_edit = bot.edit_calls[-1]["text"]
assert "cancelled" in last_edit.lower()
assert session_id in last_edit
@pytest.mark.anyio
async def test_handle_message_error_preserves_resume_token() -> None:
from takopi.bridge import BridgeConfig, handle_message
bot = _FakeBot()
session_id = "019b66fc-64c2-7a71-81cd-081c504cfeb2"
runner = ScriptRunner(
[Raise(RuntimeError("boom"))],
engine=CODEX_ENGINE,
resume_value=session_id,
)
cfg = BridgeConfig(
bot=bot,
runner=runner,
chat_id=123,
final_notify=True,
startup_msg="",
)
await handle_message(
cfg,
chat_id=123,
user_msg_id=10,
text="do something",
resume_token=None,
)
assert bot.edit_calls
last_edit = bot.edit_calls[-1]["text"]
assert "error" in last_edit.lower()
assert session_id in last_edit
assert "codex resume" in last_edit.lower()
@pytest.mark.anyio
async def test_send_with_resume_waits_for_token() -> None:
from takopi.bridge import RunningTask, _send_with_resume
bot = _FakeBot()
sent: list[tuple[int, int, str, ResumeToken | None]] = []
def enqueue(chat_id: int, user_msg_id: int, text: str, resume: ResumeToken) -> None:
sent.append((chat_id, user_msg_id, text, resume))
running_task = RunningTask()
async def trigger_resume() -> None:
await anyio.sleep(0)
running_task.resume = ResumeToken(engine=CODEX_ENGINE, value="abc123")
running_task.resume_ready.set()
async with anyio.create_task_group() as tg:
tg.start_soon(trigger_resume)
await _send_with_resume(
bot,
enqueue,
running_task,
123,
10,
"hello",
)
assert sent == [
(123, 10, "hello", ResumeToken(engine=CODEX_ENGINE, value="abc123"))
]
@pytest.mark.anyio
async def test_send_with_resume_reports_when_missing() -> None:
from takopi.bridge import RunningTask, _send_with_resume
bot = _FakeBot()
sent: list[tuple[int, int, str, ResumeToken | None]] = []
def enqueue(chat_id: int, user_msg_id: int, text: str, resume: ResumeToken) -> None:
sent.append((chat_id, user_msg_id, text, resume))
running_task = RunningTask()
running_task.done.set()
await _send_with_resume(
bot,
enqueue,
running_task,
123,
10,
"hello",
)
assert sent == []
assert bot.send_calls
assert "resume token" in bot.send_calls[-1]["text"].lower()
@pytest.mark.anyio
async def test_run_main_loop_routes_reply_to_running_resume() -> None:
from takopi.bridge import BridgeConfig, _run_main_loop
progress_ready = anyio.Event()
stop_polling = anyio.Event()
reply_ready = anyio.Event()
hold = anyio.Event()
class _BotWithProgress(_FakeBot):
def __init__(self) -> None:
super().__init__()
self.progress_id: int | None = None
async def send_message(
self,
chat_id: int,
text: str,
reply_to_message_id: int | None = None,
disable_notification: bool | None = False,
entities: list[dict] | None = None,
parse_mode: str | None = None,
) -> dict:
msg = await super().send_message(
chat_id=chat_id,
text=text,
reply_to_message_id=reply_to_message_id,
disable_notification=disable_notification,
entities=entities,
parse_mode=parse_mode,
)
if self.progress_id is None and reply_to_message_id is not None:
self.progress_id = int(msg["message_id"])
progress_ready.set()
return msg
bot = _BotWithProgress()
resume_value = "abc123"
runner = ScriptRunner(
[Wait(hold), Sleep(0.05), Return(answer="ok")],
engine=CODEX_ENGINE,
resume_value=resume_value,
)
cfg = BridgeConfig(
bot=bot,
runner=runner,
chat_id=123,
final_notify=True,
startup_msg="",
)
async def poller(_cfg: BridgeConfig):
yield {
"message_id": 1,
"text": "first",
"chat": {"id": 123},
"from": {"id": 123},
}
await progress_ready.wait()
assert bot.progress_id is not None
reply_ready.set()
yield {
"message_id": 2,
"text": "followup",
"chat": {"id": 123},
"from": {"id": 123},
"reply_to_message": {"message_id": bot.progress_id},
}
await stop_polling.wait()
async with anyio.create_task_group() as tg:
tg.start_soon(_run_main_loop, cfg, poller)
try:
with anyio.fail_after(2):
await reply_ready.wait()
await anyio.sleep(0)
hold.set()
with anyio.fail_after(2):
while len(runner.calls) < 2:
await anyio.sleep(0)
assert runner.calls[1][1] == ResumeToken(
engine=CODEX_ENGINE, value=resume_value
)
finally:
hold.set()
stop_polling.set()
tg.cancel_scope.cancel()
tests/test_exec_render.py
+170 -131
View File
@@ -1,146 +1,116 @@
import json
from pathlib import Path
from typing import cast
from types import SimpleNamespace
from takopi.exec_render import ExecProgressRenderer, render_event_cli, render_markdown
def _loads(lines: str) -> list[dict]:
return [json.loads(line) for line in lines.strip().splitlines() if line.strip()]
FIXTURE_PATH = Path(__file__).resolve().parent / "fixtures" / "codex.jsonl"
ALL_FORMATS_FIXTURE_PATH = (
Path(__file__).resolve().parent / "fixtures" / "codex_exec_json_all_formats.jsonl"
)
ALL_FORMATS_GOLDEN_PATH = (
Path(__file__).resolve().parent / "fixtures" / "codex_exec_json_all_formats.txt"
from takopi.markdown import render_markdown
from takopi.model import TakopiEvent
from takopi.render import ExecProgressRenderer, render_event_cli
from tests.factories import (
action_completed,
action_started,
session_started,
)
SAMPLE_STREAM = """
{"type":"thread.started","thread_id":"0199a213-81c0-7800-8aa1-bbab2a035a53"}
{"type":"turn.started"}
{"type":"item.completed","item":{"id":"item_0","type":"reasoning","text":"**Searching for README files**"}}
{"type":"item.started","item":{"id":"item_1","type":"command_execution","command":"bash -lc ls","aggregated_output":"","status":"in_progress"}}
{"type":"item.completed","item":{"id":"item_1","type":"command_execution","command":"bash -lc ls","aggregated_output":"2025-09-11\\nAGENTS.md\\nCHANGELOG.md\\ncliff.toml\\ncodex-cli\\ncodex-rs\\ndocs\\nexamples\\nflake.lock\\nflake.nix\\nLICENSE\\nnode_modules\\nNOTICE\\npackage.json\\npnpm-lock.yaml\\npnpm-workspace.yaml\\nPNPM.md\\nREADME.md\\nscripts\\nsdk\\ntmp\\n","exit_code":0,"status":"completed"}}
{"type":"item.completed","item":{"id":"item_2","type":"reasoning","text":"**Checking repository root for README**"}}
{"type":"item.completed","item":{"id":"item_3","type":"agent_message","text":"Yep — there’s a `README.md` in the repository root."}}
{"type":"turn.completed","usage":{"input_tokens":24763,"cached_input_tokens":24448,"output_tokens":122}}
"""
def _format_resume(token) -> str:
return f"`codex resume {token.value}`"
def test_render_event_cli_sample_stream() -> None:
last_turn = None
SAMPLE_EVENTS: list[TakopiEvent] = [
session_started("codex", "0199a213-81c0-7800-8aa1-bbab2a035a53", title="Codex"),
action_started("a-1", "command", "bash -lc ls"),
action_completed(
"a-1",
"command",
"bash -lc ls",
ok=True,
detail={"exit_code": 0},
),
action_completed("a-2", "note", "Checking repository root for README", ok=True),
]
def test_render_event_cli_sample_events() -> None:
out: list[str] = []
for evt in _loads(SAMPLE_STREAM):
last_turn, lines = render_event_cli(evt, last_turn)
out.extend(lines)
for evt in SAMPLE_EVENTS:
out.extend(render_event_cli(evt))
assert out == [
"thread started",
"turn started",
"0. **Searching for README files**",
"1. ▸ `bash -lc ls`",
"1. ✓ `bash -lc ls`",
"2. **Checking repository root for README**",
"assistant:",
" Yep — there’s a `README.md` in the repository root.",
"turn completed",
"codex",
"▸ `bash -lc ls`",
"✓ `bash -lc ls`",
"✓ Checking repository root for README",
]
def test_render_event_cli_real_run_fixture() -> None:
events = _loads(FIXTURE_PATH.read_text(encoding="utf-8"))
last_turn = None
def test_render_event_cli_handles_action_kinds() -> None:
events: list[TakopiEvent] = [
action_completed(
"c-1", "command", "pytest -q", ok=False, detail={"exit_code": 1}
),
action_completed(
"s-1",
"web_search",
"python jsonlines parser handle unknown fields",
ok=True,
),
action_completed("t-1", "tool", "github.search_issues", ok=True),
action_completed(
"f-1",
"file_change",
"2 files",
ok=True,
detail={
"changes": [
{"path": "README.md", "kind": "add"},
{"path": "src/compute_answer.py", "kind": "update"},
]
},
),
action_completed("n-1", "note", "stream error", ok=False),
]
out: list[str] = []
for evt in events:
last_turn, lines = render_event_cli(evt, last_turn)
out.extend(lines)
out.extend(render_event_cli(evt))
print("\n".join(out))
assert out[0] == "thread started"
assert "turn started" in out
assert any(line.startswith("0. ▸ `") for line in out)
assert any(line.startswith("0. ✓ `") for line in out)
assert "assistant:" in out
assert any("takopi" in line for line in out)
assert out[-1] == "turn completed"
def test_render_event_cli_all_formats_fixture() -> None:
events = _loads(ALL_FORMATS_FIXTURE_PATH.read_text(encoding="utf-8"))
last_turn = None
out: list[str] = []
for evt in events:
last_turn, lines = render_event_cli(evt, last_turn)
out.extend(lines)
assert "thread started" in out
assert "turn started" in out
assert any(line.startswith("stream error:") for line in out)
assert any(line.startswith("4. ▸ `pytest -q`") for line in out)
assert any("✗ `pytest -q` (exit 1)" in line for line in out)
assert any(line.startswith("✗ `pytest -q` (exit 1)") for line in out)
assert any(
"searched: python jsonlines parser handle unknown fields" in line
for line in out
)
assert any("tool: github.search_issues" in line for line in out)
assert any("updated `src/compute_answer.py`" in line for line in out)
assert any(
line.startswith(
"turn failed: Aborted: required dependency `npm` is missing; cannot continue."
)
for line in out
)
assert "assistant:" in out
assert any("Legacy schema example" in line for line in out)
def test_render_event_cli_all_formats_golden() -> None:
events = _loads(ALL_FORMATS_FIXTURE_PATH.read_text(encoding="utf-8"))
last_turn = None
out: list[str] = []
for evt in events:
last_turn, lines = render_event_cli(evt, last_turn)
out.extend(lines)
print("\n".join(out))
expected = ALL_FORMATS_GOLDEN_PATH.read_text(encoding="utf-8").rstrip("\n")
assert "\n".join(out) == expected
assert any("files: +README.md, ~src/compute_answer.py" in line for line in out)
assert any(line.startswith("✗ stream error") for line in out)
def test_progress_renderer_renders_progress_and_final() -> None:
r = ExecProgressRenderer(max_actions=5)
for evt in _loads(SAMPLE_STREAM):
r = ExecProgressRenderer(max_actions=5, resume_formatter=_format_resume)
for evt in SAMPLE_EVENTS:
r.note_event(evt)
progress = r.render_progress(3.0)
assert progress.startswith("working · 3s · step 3")
assert "1\\. ✓ `bash -lc ls`" in progress
assert "resume: `0199a213-81c0-7800-8aa1-bbab2a035a53`" in progress
assert progress.startswith("working · 3s · step 2")
assert "✓ `bash -lc ls`" in progress
assert "`codex resume 0199a213-81c0-7800-8aa1-bbab2a035a53`" in progress
final = r.render_final(3.0, "answer", status="done")
assert final.startswith("done · 3s · step 3")
assert "running:" not in final
assert "ran:" not in final
assert final.startswith("done · 3s · step 2")
assert "answer" in final
assert final.rstrip().endswith("resume: `0199a213-81c0-7800-8aa1-bbab2a035a53`")
assert final.rstrip().endswith(
"`codex resume 0199a213-81c0-7800-8aa1-bbab2a035a53`"
)
def test_progress_renderer_clamps_actions_and_ignores_unknown() -> None:
r = ExecProgressRenderer(max_actions=3, command_width=20)
events = [
{
"type": "item.completed",
"item": {
"id": f"item_{i}",
"type": "command_execution",
"command": f"echo {i}",
"exit_code": 0,
"status": "completed",
},
}
action_completed(
f"item_{i}",
"command",
f"echo {i}",
ok=True,
detail={"exit_code": 0},
)
for i in range(6)
]
@@ -148,33 +118,102 @@ def test_progress_renderer_clamps_actions_and_ignores_unknown() -> None:
assert r.note_event(evt) is True
assert len(r.recent_actions) == 3
assert r.recent_actions[0].startswith("3\\. ")
assert r.recent_actions[-1].startswith("5\\. ")
assert r.note_event({"type": "mystery"}) is False
assert "echo 3" in r.recent_actions[0]
assert "echo 5" in r.recent_actions[-1]
mystery = SimpleNamespace(type="mystery")
assert r.note_event(cast(TakopiEvent, mystery)) is False
def test_progress_renderer_preserves_item_ids_in_telegram_text() -> None:
def test_progress_renderer_renders_commands_in_markdown() -> None:
r = ExecProgressRenderer(max_actions=5, command_width=None)
for i in (30, 31, 32):
r.note_event(
{
"type": "item.completed",
"item": {
"id": f"item_{i}",
"type": "command_execution",
"command": f"echo {i}",
"exit_code": 0,
"status": "completed",
},
}
action_completed(
f"item_{i}",
"command",
f"echo {i}",
ok=True,
detail={"exit_code": 0},
)
)
md = r.render_progress(0.0)
assert "30\\." in md
assert "31\\." in md
assert "32\\." in md
text, _ = render_markdown(md)
assert "30. ✓ echo 30" in text
assert "31. ✓ echo 31" in text
assert "32. ✓ echo 32" in text
assert "✓ echo 30" in text
assert "✓ echo 31" in text
assert "✓ echo 32" in text
def test_progress_renderer_handles_duplicate_action_ids() -> None:
r = ExecProgressRenderer(max_actions=5)
events = [
action_started("dup", "command", "echo first"),
action_completed(
"dup",
"command",
"echo first",
ok=True,
detail={"exit_code": 0},
),
action_started("dup", "command", "echo second"),
action_completed(
"dup",
"command",
"echo second",
ok=True,
detail={"exit_code": 0},
),
]
for evt in events:
assert r.note_event(evt) is True
assert len(r.recent_actions) == 2
assert r.recent_actions[0].startswith("✓ ")
assert "echo first" in r.recent_actions[0]
assert r.recent_actions[1].startswith("✓ ")
assert "echo second" in r.recent_actions[1]
def test_progress_renderer_collapses_action_updates() -> None:
r = ExecProgressRenderer(max_actions=5)
events = [
action_started("a-1", "command", "echo one"),
action_started("a-1", "command", "echo two"),
action_completed(
"a-1",
"command",
"echo two",
ok=True,
detail={"exit_code": 0},
),
]
for evt in events:
assert r.note_event(evt) is True
assert r.action_count == 1
assert len(r.recent_actions) == 1
assert r.recent_actions[0].startswith("✓ ")
assert "echo two" in r.recent_actions[0]
def test_progress_renderer_deterministic_output() -> None:
events = [
action_started("a-1", "command", "echo ok"),
action_completed(
"a-1",
"command",
"echo ok",
ok=True,
detail={"exit_code": 0},
),
]
r1 = ExecProgressRenderer(max_actions=5)
r2 = ExecProgressRenderer(max_actions=5)
for evt in events:
r1.note_event(evt)
r2.note_event(evt)
assert r1.render_progress(1.0) == r2.render_progress(1.0)
tests/test_exec_runner.py
+263 -47
View File
@@ -1,88 +1,304 @@
import anyio
import pytest
from takopi.exec_bridge import CodexExecRunner, EventCallback
from collections.abc import AsyncIterator
from takopi.model import (
ActionEvent,
CompletedEvent,
EngineId,
ResumeToken,
StartedEvent,
TakopiEvent,
)
from takopi.runners.codex import CodexRunner
CODEX_ENGINE = EngineId("codex")
@pytest.mark.anyio
async def test_run_serialized_serializes_same_session() -> None:
runner = CodexExecRunner(codex_cmd="codex", extra_args=[])
async def test_run_serializes_same_session() -> None:
runner = CodexRunner(codex_cmd="codex", extra_args=[])
gate = anyio.Event()
in_flight = 0
max_in_flight = 0
async def run_stub(*_args, **_kwargs):
async def run_stub(*_args, **_kwargs) -> AsyncIterator[TakopiEvent]:
nonlocal in_flight, max_in_flight
in_flight += 1
max_in_flight = max(max_in_flight, in_flight)
await gate.wait()
in_flight -= 1
return ("sid", "ok", True)
try:
await gate.wait()
yield CompletedEvent(
engine=CODEX_ENGINE,
resume=ResumeToken(engine=CODEX_ENGINE, value="sid"),
ok=True,
answer="ok",
)
finally:
in_flight -= 1
runner.run = run_stub # type: ignore[assignment]
runner._run = run_stub # type: ignore[assignment]
async def drain(prompt: str, resume: ResumeToken | None) -> None:
async for _event in runner.run(prompt, resume):
pass
token = ResumeToken(engine=CODEX_ENGINE, value="sid")
async with anyio.create_task_group() as tg:
tg.start_soon(runner.run_serialized, "a", "sid")
tg.start_soon(runner.run_serialized, "b", "sid")
tg.start_soon(drain, "a", token)
tg.start_soon(drain, "b", token)
await anyio.sleep(0)
gate.set()
assert max_in_flight == 1
@pytest.mark.anyio
async def test_run_serialized_allows_parallel_new_sessions() -> None:
runner = CodexExecRunner(codex_cmd="codex", extra_args=[])
async def test_run_allows_parallel_new_sessions() -> None:
runner = CodexRunner(codex_cmd="codex", extra_args=[])
gate = anyio.Event()
in_flight = 0
max_in_flight = 0
async def run_stub(*_args, **_kwargs):
async def run_stub(*_args, **_kwargs) -> AsyncIterator[TakopiEvent]:
nonlocal in_flight, max_in_flight
in_flight += 1
max_in_flight = max(max_in_flight, in_flight)
await gate.wait()
in_flight -= 1
return ("sid", "ok", True)
try:
await gate.wait()
yield CompletedEvent(
engine=CODEX_ENGINE,
resume=ResumeToken(engine=CODEX_ENGINE, value="sid"),
ok=True,
answer="ok",
)
finally:
in_flight -= 1
runner.run = run_stub # type: ignore[assignment]
runner._run = run_stub # type: ignore[assignment]
async def drain(prompt: str, resume: ResumeToken | None) -> None:
async for _event in runner.run(prompt, resume):
pass
async with anyio.create_task_group() as tg:
tg.start_soon(runner.run_serialized, "a", None)
tg.start_soon(runner.run_serialized, "b", None)
with anyio.move_on_after(1):
while max_in_flight < 2:
await anyio.sleep(0)
tg.start_soon(drain, "a", None)
tg.start_soon(drain, "b", None)
await anyio.sleep(0)
gate.set()
assert max_in_flight == 2
@pytest.mark.anyio
async def test_new_session_holds_lock_for_resumes() -> None:
runner = CodexExecRunner(codex_cmd="codex", extra_args=[])
finish = anyio.Event()
resume_started = anyio.Event()
async def test_run_allows_parallel_different_sessions() -> None:
runner = CodexRunner(codex_cmd="codex", extra_args=[])
gate = anyio.Event()
in_flight = 0
max_in_flight = 0
async def run_stub(
_prompt: str,
session_id: str | None,
on_event: EventCallback | None = None,
) -> tuple[str, str, bool]:
if session_id is None:
if on_event:
await on_event({"type": "thread.started", "thread_id": "sid"})
await finish.wait()
return ("sid", "ok", True)
resume_started.set()
return ("sid", "ok", True)
async def run_stub(*_args, **_kwargs) -> AsyncIterator[TakopiEvent]:
nonlocal in_flight, max_in_flight
in_flight += 1
max_in_flight = max(max_in_flight, in_flight)
try:
await gate.wait()
yield CompletedEvent(
engine=CODEX_ENGINE,
resume=ResumeToken(engine=CODEX_ENGINE, value="sid"),
ok=True,
answer="ok",
)
finally:
in_flight -= 1
runner.run = run_stub # type: ignore[assignment]
runner._run = run_stub # type: ignore[assignment]
async def drain(prompt: str, resume: ResumeToken | None) -> None:
async for _event in runner.run(prompt, resume):
pass
token_a = ResumeToken(engine=CODEX_ENGINE, value="sid-a")
token_b = ResumeToken(engine=CODEX_ENGINE, value="sid-b")
async with anyio.create_task_group() as tg:
tg.start_soon(drain, "a", token_a)
tg.start_soon(drain, "b", token_b)
await anyio.sleep(0)
gate.set()
assert max_in_flight == 2
@pytest.mark.anyio
async def test_run_serializes_new_session_after_session_is_known(
tmp_path, monkeypatch
) -> None:
gate_path = tmp_path / "gate"
resume_marker = tmp_path / "resume_started"
thread_id = "019b73c4-0c3f-7701-a0bb-aac6b4d8a3bc"
codex_path = tmp_path / "codex"
codex_path.write_text(
"#!/usr/bin/env python3\n"
"import json\n"
"import os\n"
"import sys\n"
"import time\n"
"\n"
"gate = os.environ['CODEX_TEST_GATE']\n"
"resume_marker = os.environ['CODEX_TEST_RESUME_MARKER']\n"
"thread_id = os.environ['CODEX_TEST_THREAD_ID']\n"
"\n"
"args = sys.argv[1:]\n"
"if 'resume' in args:\n"
" print(json.dumps({'type': 'thread.started', 'thread_id': thread_id}), flush=True)\n"
" with open(resume_marker, 'w', encoding='utf-8') as f:\n"
" f.write('started')\n"
" f.flush()\n"
" sys.exit(0)\n"
"\n"
"print(json.dumps({'type': 'thread.started', 'thread_id': thread_id}), flush=True)\n"
"while not os.path.exists(gate):\n"
" time.sleep(0.001)\n"
"sys.exit(0)\n",
encoding="utf-8",
)
codex_path.chmod(0o755)
monkeypatch.setenv("CODEX_TEST_GATE", str(gate_path))
monkeypatch.setenv("CODEX_TEST_RESUME_MARKER", str(resume_marker))
monkeypatch.setenv("CODEX_TEST_THREAD_ID", thread_id)
runner = CodexRunner(codex_cmd=str(codex_path), extra_args=[])
session_started = anyio.Event()
resume_value: str | None = None
new_done = anyio.Event()
async def run_new() -> None:
nonlocal resume_value
async for event in runner.run("hello", None):
if isinstance(event, StartedEvent):
resume_value = event.resume.value
session_started.set()
new_done.set()
async def run_resume() -> None:
assert resume_value is not None
async for _event in runner.run(
"resume", ResumeToken(engine=CODEX_ENGINE, value=resume_value)
):
pass
async with anyio.create_task_group() as tg:
tg.start_soon(runner.run_serialized, "first", None)
await anyio.sleep(0)
tg.start_soon(runner.run_serialized, "resume", "sid")
await anyio.sleep(0)
assert not resume_started.is_set()
finish.set()
tg.start_soon(run_new)
await session_started.wait()
tg.start_soon(run_resume)
await anyio.sleep(0.01)
assert not resume_marker.exists()
gate_path.write_text("go", encoding="utf-8")
await new_done.wait()
with anyio.fail_after(2):
while not resume_marker.exists():
await anyio.sleep(0.001)
@pytest.mark.anyio
async def test_codex_runner_preserves_warning_order(tmp_path) -> None:
thread_id = "019b73c4-0c3f-7701-a0bb-aac6b4d8a3bc"
codex_path = tmp_path / "codex"
codex_path.write_text(
"#!/usr/bin/env python3\n"
"import json\n"
"import sys\n"
"\n"
"sys.stdin.read()\n"
"print(json.dumps({'type': 'error', 'message': 'warning one', 'fatal': False}), flush=True)\n"
f"print(json.dumps({{'type': 'thread.started', 'thread_id': '{thread_id}'}}), flush=True)\n"
"print(json.dumps({'type': 'item.completed', 'item': {'id': 'item_0', 'type': 'agent_message', 'text': 'ok'}}), flush=True)\n",
encoding="utf-8",
)
codex_path.chmod(0o755)
runner = CodexRunner(codex_cmd=str(codex_path), extra_args=[])
seen = [evt async for evt in runner.run("hi", None)]
assert len(seen) == 3
assert isinstance(seen[0], ActionEvent)
assert seen[0].phase == "completed"
assert seen[0].ok is False
assert seen[0].action.kind == "warning"
assert seen[0].action.title == "warning one"
assert isinstance(seen[1], StartedEvent)
assert seen[1].resume.value == thread_id
assert isinstance(seen[2], CompletedEvent)
assert seen[2].resume == seen[1].resume
assert seen[2].answer == "ok"
@pytest.mark.anyio
async def test_run_serializes_two_new_sessions_same_thread(
tmp_path, monkeypatch
) -> None:
gate_path = tmp_path / "gate"
thread_id = "019b73c4-0c3f-7701-a0bb-aac6b4d8a3bc"
codex_path = tmp_path / "codex"
codex_path.write_text(
"#!/usr/bin/env python3\n"
"import json\n"
"import os\n"
"import sys\n"
"import time\n"
"\n"
"gate = os.environ['CODEX_TEST_GATE']\n"
"thread_id = os.environ['CODEX_TEST_THREAD_ID']\n"
"\n"
"print(json.dumps({'type': 'thread.started', 'thread_id': thread_id}), flush=True)\n"
"while not os.path.exists(gate):\n"
" time.sleep(0.001)\n"
"sys.exit(0)\n",
encoding="utf-8",
)
codex_path.chmod(0o755)
monkeypatch.setenv("CODEX_TEST_GATE", str(gate_path))
monkeypatch.setenv("CODEX_TEST_THREAD_ID", thread_id)
runner = CodexRunner(codex_cmd=str(codex_path), extra_args=[])
started_first = anyio.Event()
started_second = anyio.Event()
async def run_first() -> None:
async for event in runner.run("one", None):
if isinstance(event, StartedEvent):
started_first.set()
async def run_second() -> None:
async for event in runner.run("two", None):
if isinstance(event, StartedEvent):
started_second.set()
async with anyio.create_task_group() as tg:
tg.start_soon(run_first)
tg.start_soon(run_second)
with anyio.fail_after(2):
while not (started_first.is_set() or started_second.is_set()):
await anyio.sleep(0.001)
assert not (started_first.is_set() and started_second.is_set())
gate_path.write_text("go", encoding="utf-8")
with anyio.fail_after(2):
await started_first.wait()
await started_second.wait()
tests/test_onboarding.py
+17 -11
View File
@@ -2,46 +2,52 @@ from __future__ import annotations
from pathlib import Path
from takopi import onboarding
from takopi import engines, onboarding
def test_check_setup_marks_missing_codex(monkeypatch, tmp_path: Path) -> None:
monkeypatch.setattr(onboarding.shutil, "which", lambda _name: None)
backend = engines.get_backend("codex")
monkeypatch.setattr(engines.shutil, "which", lambda _name: None)
monkeypatch.setattr(
onboarding,
"load_telegram_config",
lambda: ({"bot_token": "token", "chat_id": 123}, tmp_path / "takopi.toml"),
)
result = onboarding.check_setup()
result = onboarding.check_setup(backend)
assert result.missing_codex is True
assert result.missing_or_invalid_config is False
titles = {issue.title for issue in result.issues}
assert "Install the Codex CLI" in titles
assert "Create a config" not in titles
assert result.ok is False
def test_check_setup_marks_missing_config(monkeypatch) -> None:
monkeypatch.setattr(onboarding.shutil, "which", lambda _name: "/usr/bin/codex")
backend = engines.get_backend("codex")
monkeypatch.setattr(engines.shutil, "which", lambda _name: "/usr/bin/codex")
def _raise() -> None:
raise onboarding.ConfigError("Missing config file")
monkeypatch.setattr(onboarding, "load_telegram_config", _raise)
result = onboarding.check_setup()
result = onboarding.check_setup(backend)
assert result.missing_or_invalid_config is True
titles = {issue.title for issue in result.issues}
assert "Create a config" in titles
assert result.config_path == onboarding.HOME_CONFIG_PATH
def test_check_setup_marks_invalid_chat_id(monkeypatch, tmp_path: Path) -> None:
monkeypatch.setattr(onboarding.shutil, "which", lambda _name: "/usr/bin/codex")
backend = engines.get_backend("codex")
monkeypatch.setattr(engines.shutil, "which", lambda _name: "/usr/bin/codex")
monkeypatch.setattr(
onboarding,
"load_telegram_config",
lambda: ({"bot_token": "token", "chat_id": "123"}, tmp_path / "takopi.toml"),
)
result = onboarding.check_setup()
result = onboarding.check_setup(backend)
assert result.missing_or_invalid_config is True
titles = {issue.title for issue in result.issues}
assert "Create a config" in titles
tests/test_rendering.py
+1 -1
View File
@@ -1,4 +1,4 @@
from takopi.exec_render import render_markdown
from takopi.markdown import render_markdown
def test_render_markdown_basic_entities() -> None:
tests/test_runner_contract.py
+106
View File
@@ -0,0 +1,106 @@
import anyio
import pytest
from collections.abc import AsyncGenerator
from typing import cast
from takopi.model import (
Action,
ActionEvent,
CompletedEvent,
EngineId,
ResumeToken,
StartedEvent,
TakopiEvent,
)
from takopi.runners.mock import Emit, Return, ScriptRunner, Wait
from tests.factories import action_started
CODEX_ENGINE = EngineId("codex")
@pytest.mark.anyio
async def test_runner_contract_session_started_and_order() -> None:
raw_completed: TakopiEvent = ActionEvent(
engine=CODEX_ENGINE,
action=Action(
id="a-1",
kind="command",
title="echo ok",
detail={"exit_code": 0},
),
phase="completed",
)
script = [
Emit(action_started("a-1", "command", "echo ok")),
Emit(raw_completed),
Return(answer="done"),
]
runner = ScriptRunner(script, engine=CODEX_ENGINE, resume_value="abc123")
seen = [evt async for evt in runner.run("hi", None)]
session_events = [evt for evt in seen if isinstance(evt, StartedEvent)]
assert len(session_events) == 1
completed_events = [evt for evt in seen if isinstance(evt, CompletedEvent)]
assert len(completed_events) == 1
assert seen[-1].type == "completed"
session_idx = seen.index(session_events[0])
completed_idx = seen.index(completed_events[0])
assert session_idx < completed_idx
assert completed_events[0].resume == session_events[0].resume
assert completed_events[0].answer == "done"
assert [evt.type for evt in seen if evt.type not in {"started", "completed"}] == [
"action",
"action",
]
completed_event = next(
evt for evt in seen if isinstance(evt, ActionEvent) and evt.phase == "completed"
)
assert completed_event.type == "action"
assert completed_event.ok is True
action = completed_event.action
assert action.id == "a-1"
assert action.kind == "command"
assert action.title == "echo ok"
@pytest.mark.anyio
async def test_runner_contract_resume_matches_session_started() -> None:
runner = ScriptRunner(
[Return(answer="ok")], engine=CODEX_ENGINE, resume_value="sid"
)
seen = [evt async for evt in runner.run("hello", None)]
session = next(evt for evt in seen if isinstance(evt, StartedEvent))
completed = next(evt for evt in seen if isinstance(evt, CompletedEvent))
assert completed.resume == session.resume
assert isinstance(completed.resume, ResumeToken)
@pytest.mark.anyio
async def test_runner_releases_lock_when_consumer_closes() -> None:
gate = anyio.Event()
runner = ScriptRunner([Wait(gate)], engine=CODEX_ENGINE, resume_value="sid")
gen = cast(AsyncGenerator[TakopiEvent, None], runner.run("hello", None))
try:
while True:
evt = await anext(gen)
if isinstance(evt, StartedEvent):
break
finally:
await gen.aclose()
gen2 = cast(
AsyncGenerator[TakopiEvent, None],
runner.run("again", ResumeToken(engine=CODEX_ENGINE, value="sid")),
)
try:
while True:
evt2 = await anext(gen2)
if isinstance(evt2, StartedEvent):
break
finally:
await gen2.aclose()
tests/test_subprocess.py
+11 -4
View File
@@ -2,16 +2,23 @@ import sys
import pytest
from takopi import exec_bridge
from takopi.runners import codex
@pytest.mark.anyio
async def test_manage_subprocess_kills_when_terminate_times_out() -> None:
async with exec_bridge.manage_subprocess(
async def test_manage_subprocess_kills_when_terminate_times_out(
monkeypatch,
) -> None:
async def fake_wait_for_process(_proc, timeout: float) -> bool:
_ = timeout
return True
monkeypatch.setattr(codex, "_wait_for_process", fake_wait_for_process)
async with codex.manage_subprocess(
sys.executable,
"-c",
"import signal, time; signal.signal(signal.SIGTERM, signal.SIG_IGN); time.sleep(10)",
terminate_timeout=0.01,
) as proc:
assert proc.returncode is None