feat(cli): stream live call events + build logs (crew observability, phase 1)

README.md

@@ -116,12 +116,20 @@ deps in the pyproject).
 ## CLI
 
 ```sh
-smallestai auth login               # store your API key
-smallestai agents list              # list, get, call, and manage agents
-smallestai agent-crew deploy ...     # package and deploy crew code
-smallestai agent-crew chat           # talk to a running crew locally
+smallestai auth login                        # store your API key
+smallestai agents list                       # list, get, call, and manage agents
+smallestai calls list                        # inspect call logs, transcripts, recordings
+smallestai calls events <call-id>            # stream a live call's events (transcript, latency, tools)
+smallestai calls transcript <call-id> -f     # stream the transcript live
+smallestai models                            # text-to-speech, speech-to-text, voices
+smallestai agent-crew deploy ...             # package and deploy crew code
+smallestai agent-crew logs [build-id]        # stream a build's compile + deploy logs
+smallestai agent-crew chat                   # talk to a running crew locally
 ```
 
+> The speech command group is now `models` (text-to-speech, speech-to-text,
+> voices); `waves` still works as a hidden, back-compatible alias.
+
 ## Async client
 
 The SDK exports an `async` client with the same surface:

docs/guides/calls.md

@@ -33,6 +33,19 @@ smallestai calls transcript CALL-...
 smallestai calls recording CALL-...
 ```
 
+## Live events (while a call is in progress)
+
+Stream a call's events as they happen — transcript turns, per-turn latency, node
+transitions, tool calls, and errors:
+
+```bash
+smallestai calls events CALL-...               # full live event stream
+smallestai calls transcript CALL-... --follow  # just the conversation, live
+```
+
+The call must be in progress; for a finished call use `calls transcript` (no
+`--follow`) or `calls get`. Add `--json` to emit one raw event per line.
+
 ## Reading a transfer from the logs
 
 A transfer produces a second **transfer leg**. If that leg shows

docs/guides/crew.md

@@ -44,6 +44,7 @@ crew, and a `requirements.txt`. Deploy with:
 ```bash
 smallestai agent-crew deploy --entry-point server.py
 smallestai agent-crew builds        # watch build status; Make Live when SUCCEEDED
+smallestai agent-crew logs [build-id]   # stream a build's compile + deploy logs (latest by default)
 ```
 
 Pin the SDK in `requirements.txt`:

src/smallestai/cli/agent_crew.py

@@ -476,70 +476,63 @@ async def _manage_build(agent_id: str, build, access_token: str):
                 )
                 console.print(f"[bold green]✓ Build {build.id[:12]}... has been taken down.[/bold green]")
 
-    # @app.command("logs")
-    # def stream_build(
-    #     build_id: str = typer.Argument(..., help="The build ID to stream logs for"),
-    # ):
-    #     """
-    #     Stream build logs in real-time using Server-Sent Events.
-    #     """
-    #     asyncio.run(async_stream_build(build_id))
-
-    # async def async_stream_build(build_id: str):
-    #     """Async implementation of stream build command."""
-    #     agent_id = project_config.get_agent_id()
-
-    #     if not agent_id:
-    #         console.print(
-    #             "[red]Agent not initialized. Run 'smallestai agent init' first.[/red]"
-    #         )
-    #         return
-
-    #     credentials = auth_client.get_credentials()
-    #     if not credentials or not credentials.get("access_token"):
-    #         console.print(
-    #             "[red]Error: You must be logged in first. Run 'smallestai auth login'[/red]"
-    #         )
-    #         raise typer.Exit(1)
-
-    #     access_token = credentials["access_token"]
-
-    #     console.print(f"[bold cyan]Streaming logs for build: {build_id}[/bold cyan]")
-    #     console.print("[dim]Press Ctrl+C to stop streaming[/dim]\n")
-
-    #     try:
-    #         async for event in atoms_client.stream_agent_build(
-    #             agent_id=agent_id,
-    #             build_id=build_id,
-    #             api_key=access_token,
-    #         ):
-    #             event_type = event.get("type")
-
-    #             if event_type == "log":
-    #                 console.print(f"[dim]LOG:[/dim] {event.get('message', '')}")
-    #             elif event_type == "status":
-    #                 status = event.get("status", "")
-    #                 status_style = {
-    #                     "SUCCEEDED": "[green]SUCCEEDED[/green]",
-    #                     "BUILD_FAILED": "[red]BUILD_FAILED[/red]",
-    #                     "DEPLOY_FAILED": "[red]DEPLOY_FAILED[/red]",
-    #                     "PENDING": "[yellow]PENDING[/yellow]",
-    #                     "BUILDING": "[yellow]BUILDING[/yellow]",
-    #                     "DEPLOYING": "[yellow]DEPLOYING[/yellow]",
-    #                 }.get(status, status)
-    #                 console.print(f"[bold]STATUS:[/bold] {status_style}")
-
-    #                 if status in ["SUCCEEDED", "BUILD_FAILED", "DEPLOY_FAILED"]:
-    #                     console.print("\n[bold]Build stream ended.[/bold]")
-    #                     break
-    #             elif event_type == "error":
-    #                 console.print(f"[red]ERROR:[/red] {event.get('message', '')}")
-    #                 break
-
-    #     except KeyboardInterrupt:
-    #         console.print("\n[yellow]Streaming stopped by user.[/yellow]")
-    #     except Exception as e:
-    #         console.print(f"[red]Error streaming build logs: {e}[/red]")
+    @app.command("logs")
+    def build_logs(
+        build_id: str = typer.Argument(None, help="Build ID to stream logs for (defaults to the latest build)"),
+    ):
+        """Stream a build's logs (compile + deploy) in real time.
+
+        With no build ID, streams the most recent build for the current agent.
+        Use this to debug a deploy that failed or to watch one in progress.
+        """
+        asyncio.run(async_build_logs(build_id))
+
+    async def async_build_logs(build_id: str | None):
+        agent_id = project_config.get_agent_id()
+        if not agent_id:
+            console.print("[red]Agent not initialized. Run 'smallestai agent-crew init' first.[/red]")
+            raise typer.Exit(1)
+
+        credentials = auth_client.get_credentials()
+        if not credentials or not credentials.get("access_token"):
+            console.print("[red]Error: You must be logged in first. Run 'smallestai auth login'[/red]")
+            raise typer.Exit(1)
+        access_token = credentials["access_token"]
+
+        if not build_id:
+            result = await atoms_client.list_agent_builds(agent_id=agent_id, api_key=access_token, limit=1, offset=0)
+            if not result.builds:
+                console.print("[yellow]No builds found. Run 'smallestai agent-crew deploy' first.[/yellow]")
+                raise typer.Exit(1)
+            build_id = result.builds[0].id
+            console.print(f"[dim]Latest build: {build_id}[/dim]")
+
+        console.print(
+            f"[bold cyan]Streaming logs for build {build_id[:12]}...[/bold cyan]  [dim](Ctrl+C to stop)[/dim]\n"
+        )
+
+        terminal = {"SUCCEEDED", "BUILD_FAILED", "DEPLOY_FAILED"}
+        try:
+            async for event in atoms_client.stream_agent_build(
+                agent_id=agent_id, build_id=build_id, access_token=access_token
+            ):
+                etype = event.get("type")
+                if etype == "log":
+                    console.print(event.get("message", ""), highlight=False)
+                elif etype == "status":
+                    status = str(event.get("status", ""))
+                    color = {"SUCCEEDED": "green", "BUILD_FAILED": "red", "DEPLOY_FAILED": "red"}.get(status, "yellow")
+                    console.print(f"[bold {color}]● {status}[/bold {color}]")
+                    if status in terminal:
+                        break
+                elif etype == "error":
+                    console.print(f"[red]error:[/red] {event.get('message', '')}")
+                    break
+        except KeyboardInterrupt:
+            console.print("\n[yellow]Stopped.[/yellow]")
+        except Exception as e:
+            console.print(f"[red]Error streaming build logs: {e}[/red]")
+            raise typer.Exit(1)
 
     @app.command()
     def doctor(

src/smallestai/cli/calls.py

@@ -4,18 +4,101 @@
 `smallestai agents call`.
 """
 
+import asyncio
 import json as _json
 
+import httpx
 import typer
 from rich.console import Console
 from rich.table import Table
 
+from smallestai.cli.lib.atoms import AtomsAPIClient
 from smallestai.cli.lib.auth import AuthClient
 from smallestai.cli.lib.client import make_client
 
 console = Console()
 
 
+def _session_token(auth_client: AuthClient) -> str:
+    """Session access token for live endpoints (events SSE) that need it."""
+    creds = auth_client.get_credentials()
+    token = (creds or {}).get("access_token")
+    if not token:
+        console.print("[red]You must be logged in. Run 'smallestai auth login'.[/red]")
+        raise typer.Exit(1)
+    return token
+
+
+def _render_event(ev: dict) -> None:
+    """Pretty-print one live call event."""
+    et = ev.get("event_type")
+    if et == "call_start":
+        console.print("[dim]● call started[/dim]")
+    elif et == "user_transcription":
+        console.print(f"[cyan]user[/cyan]: {ev.get('user_transcription_text', '')}")
+    elif et == "tts_completed":
+        console.print(f"[green]agent[/green]: {ev.get('tts_text', '')}")
+    elif et == "turn_latency":
+        console.print(f"[dim]  latency: {ev.get('turn_latency')}[/dim]")
+    elif et == "agent_node_state":
+        console.print(f"[dim]  node: {ev.get('node_name')} ({ev.get('node_type')})[/dim]")
+    elif et == "hopping":
+        console.print(f"[dim]  hop: {ev.get('from_node_id')} -> {ev.get('to_node_id')}[/dim]")
+    elif et in ("tool_call_start", "tool_call_end", "tool_call_error"):
+        console.print(
+            f"[magenta]  {et}[/magenta]: {_json.dumps({k: v for k, v in ev.items() if k != 'event_type'}, default=str)}"
+        )
+    elif et in ("agent_error", "generic_call_error"):
+        console.print(f"[red]  error: {ev.get('error') or ev.get('message') or ev}[/red]")
+    else:
+        rest = {k: v for k, v in ev.items() if k != "event_type"}
+        console.print(f"[dim]  {et}{': ' + _json.dumps(rest, default=str) if rest else ''}[/dim]")
+
+
+def _render_transcript_event(ev: dict) -> None:
+    """Print only the conversation turns from a live event stream."""
+    et = ev.get("event_type")
+    if et == "user_transcription" and ev.get("user_transcription_text"):
+        console.print(f"[cyan]user[/cyan]: {ev['user_transcription_text']}")
+    elif et == "tts_completed" and ev.get("tts_text"):
+        console.print(f"[green]agent[/green]: {ev['tts_text']}")
+
+
+async def _stream_call_events(call_id: str, token: str, as_json: bool, transcript_only: bool) -> None:
+    client = AtomsAPIClient()
+    try:
+        async for ev in client.stream_call_events(call_id, token):
+            et = ev.get("event_type")
+            if et == "sse_init":
+                if not as_json and not transcript_only:
+                    console.print("[dim]connected — waiting for events…[/dim]")
+                continue
+            if as_json:
+                console.print_json(_json.dumps(ev, default=str))
+            elif transcript_only:
+                _render_transcript_event(ev)
+            else:
+                _render_event(ev)
+            if et in ("sse_close", "call_end"):
+                if not as_json and not transcript_only:
+                    console.print("[dim]— call ended —[/dim]")
+                break
+    except httpx.HTTPStatusError as e:
+        code = e.response.status_code
+        if code == 400:
+            console.print(
+                f"[yellow]This call is not live (already finished). "
+                f"Run 'smallestai calls transcript {call_id}' for the final transcript.[/yellow]"
+            )
+        elif code == 404:
+            console.print("[yellow]Call not found.[/yellow]")
+        else:
+            console.print(f"[red]Error: {e}[/red]")
+        raise typer.Exit(1)
+    except KeyboardInterrupt:
+        console.print("\n[yellow]Stopped.[/yellow]")
+
+
 def _data(resp):
     """Unwrap `.data` from an SDK response (or return the object itself)."""
     return getattr(resp, "data", resp)
@@ -93,12 +176,36 @@ def get_call(
         console.print(f"  recording     : {rec or '[dim]none[/dim]'}")
         console.print("  [dim]transcript: smallestai calls transcript " + call_id + "[/dim]")
 
+    @calls_app.command("events")
+    def events(
+        call_id: str,
+        as_json: bool = typer.Option(False, "--json", help="Emit raw JSON, one event per line"),
+    ):
+        """Stream a live call's events as they happen.
+
+        Shows transcript turns, per-turn latency, node transitions, tool calls,
+        and errors while the call is in progress. For a finished call use
+        `smallestai calls transcript`.
+        """
+        token = _session_token(auth_client)
+        asyncio.run(_stream_call_events(call_id, token, as_json, transcript_only=False))
+
     @calls_app.command("transcript")
     def transcript(
         call_id: str,
+        follow: bool = typer.Option(
+            False, "--follow", "-f", help="Stream the transcript live (call must be in progress)"
+        ),
         as_json: bool = typer.Option(False, "--json", help="Emit raw JSON"),
     ):
-        """Print a call's transcript, one turn per line."""
+        """Print a call's transcript, one turn per line.
+
+        With --follow, streams turns live while the call is in progress.
+        """
+        if follow:
+            token = _session_token(auth_client)
+            asyncio.run(_stream_call_events(call_id, token, as_json, transcript_only=True))
+            return
         d = _data(make_client(auth_client).atoms.calls.get(id=call_id))
         turns = getattr(d, "transcript", None) or []
         if as_json:

src/smallestai/cli/lib/atoms.py

@@ -1,3 +1,4 @@
+import json as _json
 from enum import Enum
 from typing import List, Optional
 
@@ -305,28 +306,46 @@ async def update_agent_build(
 
             return update_build_response.data
 
-    # async def stream_agent_build(
-    #     self,
-    #     agent_id: str,
-    #     build_id: str,
-    #     api_key: str,
-    # ):
-    #     """
-    #     Stream build logs using Server-Sent Events.
-    #     Yields tuples of (event_type, data) where event_type is 'log', 'status', or 'error'.
-    #     """
-    #     async with httpx.AsyncClient(timeout=None) as client:
-    #         async with client.stream(
-    #             "GET",
-    #             f"{self.base_url}/atoms/v1/sdk/agents/{agent_id}/builds/{build_id}/stream",
-    #             headers={
-    #                 "Authorization": f"Bearer {api_key}",
-    #             },
-    #         ) as response:
-    #             response.raise_for_status()
-    #             async for line in response.aiter_lines():
-    #                 if line.startswith("data: "):
-    #                     import json
-
-    #                     data = json.loads(line[6:])
-    #                     yield data
+    async def _stream_sse(self, url: str, access_token: str):
+        """Open an SSE stream and yield each `data:` frame as a parsed dict.
+
+        Shared by build-log and call-event streaming. Blank keep-alive lines and
+        non-JSON frames are skipped. The stream ends when the server closes it.
+        """
+        async with httpx.AsyncClient(timeout=None) as client:
+            async with client.stream(
+                "GET",
+                url,
+                headers={"Authorization": f"Bearer {access_token}"},
+            ) as response:
+                response.raise_for_status()
+                async for line in response.aiter_lines():
+                    if not line.startswith("data:"):
+                        continue
+                    payload = line[5:].lstrip()
+                    if not payload:
+                        continue
+                    try:
+                        yield _json.loads(payload)
+                    except _json.JSONDecodeError:
+                        continue
+
+    async def stream_agent_build(self, agent_id: str, build_id: str, access_token: str):
+        """Stream a build's logs (SSE). Yields dicts of the form
+        `{"type": "log"|"status"|"error", "message"|"status": ...}`.
+        """
+        url = f"{self.base_url}/atoms/v1/sdk/agents/{agent_id}/builds/{build_id}/stream"
+        async for event in self._stream_sse(url, access_token):
+            yield event
+
+    async def stream_call_events(self, call_id: str, access_token: str):
+        """Stream a live call's events (SSE). Yields dicts with an `event_type`
+        field (e.g. `user_transcription`, `tts_completed`, `turn_latency`,
+        `agent_node_state`, `tool_call_start`, `agent_error`, `call_end`).
+
+        The call must be in progress; the platform returns 400 for a completed
+        call (use `calls transcript` for finished calls).
+        """
+        url = f"{self.base_url}/atoms/v1/events?callId={call_id}"
+        async for event in self._stream_sse(url, access_token):
+            yield event