Angel is a TUI-based autonomous coding agent built on fauxtp GenServers.
1"""Angel — main entry point.
2
3Starts the fauxtp actor system in a background thread and the Textual TUI
4in the main thread. Communication between them is bridged via anyio's
5BlockingPortal.
6"""
7
8from __future__ import annotations
9
10import os
11import threading
12from typing import Any
13
14import anyio
15from anyio.from_thread import BlockingPortal
16from fauxtp import cast
17
18from angel.supervisor import start_angel
19from angel.tui import AngelApp
20
21
22class ActorBridge:
23 """Thread-safe bridge between the Textual main thread and the anyio actor thread."""
24
25 def __init__(self) -> None:
26 self.user_pid: Any = None
27 self._portal: BlockingPortal | None = None
28 self._ready = threading.Event()
29
30 def wait_ready(self) -> None:
31 self._ready.wait()
32
33 def cast_to_user(self, pid: Any, message: Any) -> None:
34 """Schedule a cast to the UserServer from Textual's event loop."""
35 if self._portal:
36 self._portal.start_task_soon(cast, pid, message)
37
38
39def run_actor_system(bridge: ActorBridge, app: AngelApp) -> None:
40 """Run the fauxtp actor system in a background thread."""
41
42 async def _main() -> None:
43 async with anyio.create_task_group() as tg:
44 async with BlockingPortal() as portal:
45 bridge._portal = portal
46
47 # UI callback: runs in the anyio thread, posts to Textual
48 async def ui_callback(event_type: str, *args: Any) -> None:
49 match event_type:
50 case "agent_status":
51 app.call_from_thread(app.on_agent_status, *args)
52 case "tool_call":
53 app.call_from_thread(app.on_tool_call, *args)
54 case "tool_result":
55 app.call_from_thread(app.on_tool_result, *args)
56 case "agent_done":
57 app.call_from_thread(app.on_agent_done, *args)
58
59 model = os.environ.get("ANGEL_MODEL", "openai/gpt-4.1")
60 project_root = os.environ.get("ANGEL_PROJECT_ROOT", ".")
61
62 pids = await start_angel(
63 tg=tg,
64 ui_callback=ui_callback,
65 model=model,
66 project_root=project_root,
67 )
68
69 bridge.user_pid = pids["user"]
70 bridge._ready.set()
71
72 await anyio.sleep_forever()
73
74 anyio.run(_main)
75
76
77def main() -> None:
78 bridge = ActorBridge()
79 app = AngelApp()
80
81 # Start actor system in background thread
82 actor_thread = threading.Thread(
83 target=run_actor_system,
84 args=(bridge, app),
85 daemon=True,
86 name="angel-actors",
87 )
88 actor_thread.start()
89
90 # Wait for actors to be ready
91 bridge.wait_ready()
92
93 # Wire up the app
94 app._user_pid = bridge.user_pid
95 app._cast = bridge.cast_to_user
96
97 # Run the TUI (blocks until quit)
98 app.run()
99
100
101if __name__ == "__main__":
102 main()