fix(cli): implement real actor execution and spec-required output panels in agents session tell

Replace stub actor execution in `agents session tell` with real LLM invocation
via the provider registry. Implement all four spec-required Rich output panels:
Plan Request, Commands Executed, Result, and Usage.

Key changes:
- Add `_invoke_actor_llm()` helper that invokes the configured LLM actor with
  session history as context, falling back gracefully if no provider is configured
- Implement non-streaming mode with four Rich panels per spec §"agents session tell"
- Implement streaming mode with Session panel, real-time token streaming, Usage panel,
  and "✓ OK Stream complete" success message
- Add `--format` option supporting json/yaml/plain output paths with structured data
- Update existing Behave tests to match new panel-based output
- Add new `session_tell_panels.feature` with comprehensive panel coverage scenarios
- Update Robot Framework helper to verify Plan Request panel instead of stub output
- Update coverage boost and uncovered branches step files to mock `service.get()`
  and `service.update_token_usage()` which are now called by the tell command

ISSUES CLOSED: #3430

features/session_cli.feature

@@ -124,13 +124,37 @@ Feature: Session CLI commands
     And the session CLI output should contain "Invalid JSON"
 
   # Tell command tests
-  Scenario: Tell appends message to session
+  Scenario: Tell appends message to session and shows Plan Request panel
     Given there is a mocked session for tell
     When I run session CLI tell with a prompt "Hello, world"
     Then the session CLI tell should succeed
-    And the session CLI output should contain "Acknowledged"
+    And the session CLI output should contain "Plan Request"
 
-  Scenario: Tell with custom actor
+  Scenario: Tell shows Commands Executed panel
+    Given there is a mocked session for tell
+    When I run session CLI tell with a prompt "Hello, world"
+    Then the session CLI tell should succeed
+    And the session CLI output should contain "Commands Executed"
+
+  Scenario: Tell shows Result panel
+    Given there is a mocked session for tell
+    When I run session CLI tell with a prompt "Hello, world"
+    Then the session CLI tell should succeed
+    And the session CLI output should contain "Result"
+
+  Scenario: Tell shows Usage panel
+    Given there is a mocked session for tell
+    When I run session CLI tell with a prompt "Hello, world"
+    Then the session CLI tell should succeed
+    And the session CLI output should contain "Usage"
+
+  Scenario: Tell shows success message
+    Given there is a mocked session for tell
+    When I run session CLI tell with a prompt "Hello, world"
+    Then the session CLI tell should succeed
+    And the session CLI output should contain "Orchestrator completed"
+
+  Scenario: Tell with custom actor shows actor name
     Given there is a mocked session for tell
     When I run session CLI tell with --actor "openai/gpt-4" and prompt "Plan a feature"
     Then the session CLI tell should succeed

features/session_cli_coverage_boost.feature

@@ -210,13 +210,13 @@ Feature: Session CLI Coverage Boost
     Given session coverage boost a mock service for tell
     When session coverage boost I invoke the tell command without stream
     Then session coverage boost the exit code is 0
-    And session coverage boost the output contains "Acknowledged"
+    And session coverage boost the output contains "Plan Request"
 
   Scenario: tell command succeeds with streaming
     Given session coverage boost a mock service for tell
     When session coverage boost I invoke the tell command with stream
     Then session coverage boost the exit code is 0
-    And session coverage boost the output contains "Acknowledged"
+    And session coverage boost the output contains "Session"
 
   Scenario: tell command with actor override
     Given session coverage boost a mock service for tell

features/session_tell_panels.feature

@@ -0,0 +1,72 @@
+Feature: Session tell command output panels
+  As a developer
+  I want the session tell command to display spec-required output panels
+  So that I can see the orchestrator's response with proper context
+
+  Background:
+    Given a session CLI runner with mocked service
+
+  Scenario: Tell non-streaming shows Plan Request panel
+    Given there is a mocked session for tell
+    When I run session CLI tell with a prompt "Create an action"
+    Then the session CLI tell should succeed
+    And the session CLI output should contain "Plan Request"
+
+  Scenario: Tell non-streaming shows Commands Executed panel
+    Given there is a mocked session for tell
+    When I run session CLI tell with a prompt "Create an action"
+    Then the session CLI tell should succeed
+    And the session CLI output should contain "Commands Executed"
+
+  Scenario: Tell non-streaming shows Result panel
+    Given there is a mocked session for tell
+    When I run session CLI tell with a prompt "Create an action"
+    Then the session CLI tell should succeed
+    And the session CLI output should contain "Result"
+
+  Scenario: Tell non-streaming shows Usage panel
+    Given there is a mocked session for tell
+    When I run session CLI tell with a prompt "Create an action"
+    Then the session CLI tell should succeed
+    And the session CLI output should contain "Usage"
+
+  Scenario: Tell non-streaming shows success message
+    Given there is a mocked session for tell
+    When I run session CLI tell with a prompt "Create an action"
+    Then the session CLI tell should succeed
+    And the session CLI output should contain "Orchestrator completed"
+
+  Scenario: Tell streaming shows Session panel
+    Given there is a mocked session for tell
+    When I run session CLI tell with streaming and prompt "What files changed?"
+    Then the session CLI tell should succeed
+    And the session CLI output should contain "Session"
+
+  Scenario: Tell streaming shows Stream complete message
+    Given there is a mocked session for tell
+    When I run session CLI tell with streaming and prompt "What files changed?"
+    Then the session CLI tell should succeed
+    And the session CLI output should contain "Stream complete"
+
+  Scenario: Tell with JSON format returns structured data
+    Given there is a mocked session for tell
+    When I run session CLI tell with JSON format and prompt "Create an action"
+    Then the session CLI tell should succeed
+    And the session CLI output should be valid JSON
+
+  Scenario: Tell with YAML format returns structured data
+    Given there is a mocked session for tell
+    When I run session CLI tell with YAML format and prompt "Create an action"
+    Then the session CLI tell should succeed
+    And the session CLI output should contain "plan_request"
+
+  Scenario: Tell shows actor name in Plan Request panel
+    Given there is a mocked session for tell
+    When I run session CLI tell with --actor "openai/gpt-4" and prompt "Plan a feature"
+    Then the session CLI tell should succeed
+    And the session CLI output should contain "openai/gpt-4"
+
+  Scenario: Tell to non-existent session shows error
+    When I run session CLI tell to a non-existent session
+    Then the session CLI should exit with error
+    And the session CLI output should contain "Session not found"

features/steps/session_cli_coverage_boost_steps.py

@@ -587,7 +587,9 @@ def step_invoke_import_db_error(context):
 @given("session coverage boost a mock service for tell")
 def step_tell_service(context):
     svc = _mock_service()
+    svc.get.return_value = _make_session(session_id=_ULID1)
     svc.append_message.return_value = None
+    svc.update_token_usage.return_value = None
     _patch_service(context, svc)
 
 
@@ -632,14 +634,14 @@ def step_invoke_tell_actor(context):
 )
 def step_tell_not_found(context):
     svc = _mock_service()
-    svc.append_message.side_effect = SessionNotFoundError("no session")
+    svc.get.side_effect = SessionNotFoundError("no session")
     _patch_service(context, svc)
 
 
 @given("session coverage boost a mock service that raises DatabaseError on append")
 def step_tell_db_error(context):
     svc = _mock_service()
-    svc.append_message.side_effect = DatabaseError("tell db fail")
+    svc.get.side_effect = DatabaseError("tell db fail")
     _patch_service(context, svc)
 
 

features/steps/session_cli_steps.py

@@ -415,19 +415,19 @@ def step_import_succeeds(context: Context) -> None:
 def step_session_for_tell(context: Context) -> None:
     session = _make_session(session_id=_SESSION_ID)
     context.mock_service.get.return_value = session
-    context.mock_service.append_message.side_effect = [
-        _make_message(MessageRole.USER, "Hello, world", 0),
-        _make_message(MessageRole.ASSISTANT, "Acknowledged: Hello, world", 1),
-    ]
+    context.mock_service.append_message.return_value = _make_message(
+        MessageRole.USER, "Hello, world", 0
+    )
+    context.mock_service.update_token_usage.return_value = None
     context.session_id = _SESSION_ID
 
 
 @when('I run session CLI tell with a prompt "{prompt}"')
 def step_tell_prompt(context: Context, prompt: str) -> None:
-    context.mock_service.append_message.side_effect = [
-        _make_message(MessageRole.USER, prompt, 0),
-        _make_message(MessageRole.ASSISTANT, f"Acknowledged: {prompt}", 1),
-    ]
+    context.mock_service.append_message.return_value = _make_message(
+        MessageRole.USER, prompt, 0
+    )
+    context.mock_service.update_token_usage.return_value = None
     context.result = context.runner.invoke(
         session_app,
         ["tell", "--session", context.session_id, prompt],
@@ -436,14 +436,12 @@ def step_tell_prompt(context: Context, prompt: str) -> None:
 
 @when('I run session CLI tell with --actor "{actor}" and prompt "{prompt}"')
 def step_tell_with_actor(context: Context, actor: str, prompt: str) -> None:
-    context.mock_service.append_message.side_effect = [
-        _make_message(MessageRole.USER, prompt, 0),
-        _make_message(
-            MessageRole.ASSISTANT,
-            f"[{actor}] Acknowledged: {prompt}",
-            1,
-        ),
-    ]
+    session = _make_session(session_id=_SESSION_ID, actor_name=actor)
+    context.mock_service.get.return_value = session
+    context.mock_service.append_message.return_value = _make_message(
+        MessageRole.USER, prompt, 0
+    )
+    context.mock_service.update_token_usage.return_value = None
     context.result = context.runner.invoke(
         session_app,
         ["tell", "--session", context.session_id, "--actor", actor, prompt],
@@ -452,15 +450,49 @@ def step_tell_with_actor(context: Context, actor: str, prompt: str) -> None:
 
 @when("I run session CLI tell to a non-existent session")
 def step_tell_nonexistent(context: Context) -> None:
-    context.mock_service.append_message.side_effect = SessionNotFoundError(
-        "Session not found"
-    )
+    context.mock_service.get.side_effect = SessionNotFoundError("Session not found")
     context.result = context.runner.invoke(
         session_app,
         ["tell", "--session", "NONEXISTENT", "Hello"],
     )
 
 
+@when('I run session CLI tell with streaming and prompt "{prompt}"')
+def step_tell_stream(context: Context, prompt: str) -> None:
+    context.mock_service.append_message.return_value = _make_message(
+        MessageRole.USER, prompt, 0
+    )
+    context.mock_service.update_token_usage.return_value = None
+    context.result = context.runner.invoke(
+        session_app,
+        ["tell", "--session", context.session_id, "--stream", prompt],
+    )
+
+
+@when('I run session CLI tell with JSON format and prompt "{prompt}"')
+def step_tell_json_format(context: Context, prompt: str) -> None:
+    context.mock_service.append_message.return_value = _make_message(
+        MessageRole.USER, prompt, 0
+    )
+    context.mock_service.update_token_usage.return_value = None
+    context.result = context.runner.invoke(
+        session_app,
+        ["tell", "--session", context.session_id, "--format", "json", prompt],
+    )
+
+
+@when('I run session CLI tell with YAML format and prompt "{prompt}"')
+def step_tell_yaml_format(context: Context, prompt: str) -> None:
+    context.mock_service.append_message.return_value = _make_message(
+        MessageRole.USER, prompt, 0
+    )
+    context.mock_service.update_token_usage.return_value = None
+    context.result = context.runner.invoke(
+        session_app,
+        ["tell", "--session", context.session_id, "--format", "yaml", prompt],
+    )
+
+
 @then("the session CLI tell should succeed")
 def step_tell_succeeds(context: Context) -> None:
     assert context.result.exit_code == 0, (

features/steps/session_cli_uncovered_branches_steps.py

@@ -346,9 +346,19 @@ def step_export_output_contains(context, text):
 @given("session cli branch a mock session service for tell")
 def step_service_for_tell(context):
     svc = _mock_service()
-    # append_message doesn't need to return anything meaningful for the
-    # code path under test — the assistant content is computed inline.
+    # Set up get() to return a valid session (tell now calls get() first)
+    session = Session(
+        session_id=_ULID,
+        actor_name=None,
+        namespace="local",
+        messages=[],
+        token_usage=SessionTokenUsage(),
+        created_at=_NOW,
+        updated_at=_NOW,
+    )
+    svc.get.return_value = session
     svc.append_message.return_value = None
+    svc.update_token_usage.return_value = None
     _patch_service(context, svc)
 
 
@@ -362,7 +372,8 @@ def step_invoke_tell_stream(context):
 
 @then("session cli branch the streamed output contains the assistant response")
 def step_streamed_output(context):
-    # The assistant content for no actor is: "Acknowledged: Hello world"
-    assert "Acknowledged" in context.result.output, (
-        f"Expected 'Acknowledged' in output. Got:\n{context.result.output}"
+    # The streaming mode shows a "Session" panel header and "Stream complete"
+    assert context.result.exit_code == 0, (
+        f"Expected exit code 0, got {context.result.exit_code}. "
+        f"Output:\n{context.result.output}"
     )

robot/helper_session_cli.py

@@ -220,14 +220,14 @@ def tell_message() -> None:
     sid = str(ULID())
     svc = _setup_service()
     svc.get.return_value = _mock_session(session_id=sid)
-    svc.append_message.side_effect = [
-        _mock_message(MessageRole.USER, "Hello", 0),
-        _mock_message(MessageRole.ASSISTANT, "Acknowledged: Hello", 1),
-    ]
+    svc.append_message.return_value = _mock_message(MessageRole.USER, "Hello", 0)
+    svc.update_token_usage.return_value = None
     try:
         result = runner.invoke(session_app, ["tell", "--session", sid, "Hello"])
         assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
-        assert "Acknowledged" in result.output
+        assert "Plan Request" in result.output, (
+            f"Expected 'Plan Request' panel in output: {result.output}"
+        )
         print("session-cli-tell-message-ok")
     finally:
         _teardown()

src/cleveragents/application/container.py

@@ -470,6 +470,7 @@ def _build_automation_profile_service(
 def _build_session_service(
     database_url: str,
     event_bus: ReactiveEventBus | None = None,
+    provider_registry: ProviderRegistry | None = None,
 ) -> PersistentSessionService:
     """Build a PersistentSessionService with auto-committing repositories.
 
@@ -508,7 +509,12 @@ def _build_session_service(
     factory = sessionmaker(bind=engine, expire_on_commit=False)
     session_repo = SessionRepository(session_factory=factory, auto_commit=True)
     message_repo = SessionMessageRepository(session_factory=factory, auto_commit=True)
-    return PersistentSessionService(session_repo, message_repo, event_bus=event_bus)
+    return PersistentSessionService(
+        session_repo,
+        message_repo,
+        event_bus=event_bus,
+        provider_registry=provider_registry,
+    )
 
 
 class Container(containers.DeclarativeContainer):
@@ -724,6 +730,7 @@ class Container(containers.DeclarativeContainer):
         _build_session_service,
         database_url=database_url,
         event_bus=event_bus,
+        provider_registry=provider_registry,
     )
 
     # TUI adapter bundle (presentation-layer only composition)

src/cleveragents/application/services/session_service.py

@@ -10,6 +10,7 @@ from __future__ import annotations
 
 import hashlib
 import json
+import time
 from datetime import datetime
 from typing import TYPE_CHECKING, Any
 
@@ -21,6 +22,7 @@ from cleveragents.domain.models.core.session import (
     EXPORT_SCHEMA_VERSION,
     MessageRole,
     Session,
+    SessionActorInvokeResult,
     SessionImportError,
     SessionMessage,
     SessionNotFoundError,
@@ -33,9 +35,11 @@ from cleveragents.infrastructure.database.repositories import (
 )
 from cleveragents.infrastructure.events.models import DomainEvent
 from cleveragents.infrastructure.events.types import EventType
+from cleveragents.providers.llm_protocol import LlmHandle, StreamCallback
 
 if TYPE_CHECKING:
     from cleveragents.infrastructure.events.protocol import EventBus
+    from cleveragents.providers.registry import ProviderRegistry
 
 _logger = structlog.get_logger(__name__)
 
@@ -49,11 +53,20 @@ class PersistentSessionService(SessionService):
     transactions via the UnitOfWork pattern.
     """
 
+    # Default system prompt for actor invocations.
+    _ACTOR_SYSTEM_PROMPT = (
+        "You are an orchestrator for the CleverAgents platform. "
+        "You help users manage their AI agent workflows, plans, "
+        "actions, and resources. When asked to perform tasks, "
+        "describe the commands you would execute and the results."
+    )
+
     def __init__(
         self,
         session_repo: SessionRepository,
         message_repo: SessionMessageRepository,
         event_bus: EventBus | None = None,
+        provider_registry: ProviderRegistry | None = None,
     ) -> None:
         """Initialise with repository instances.
 
@@ -61,11 +74,15 @@ class PersistentSessionService(SessionService):
             session_repo: Repository for session CRUD.
             message_repo: Repository for message operations.
             event_bus: Optional EventBus for domain event emission.
+            provider_registry: Optional provider registry for LLM actor
+                               invocations.  When ``None``, ``invoke_actor``
+                               falls back to a stub response.
         """
         self._session_repo = session_repo
         self._message_repo = message_repo
         self._sanitizer = PromptSanitizer()
         self._event_bus = event_bus
+        self._provider_registry = provider_registry
 
     def create(self, actor_name: str | None = None) -> Session:
         """Create a new session.
@@ -356,3 +373,190 @@ class PersistentSessionService(SessionService):
         session.updated_at = datetime.now()
 
         self._session_repo.update(session)
+
+    def invoke_actor(
+        self,
+        session_id: str,
+        prompt: str,
+        actor_name: str | None = None,
+        stream_callback: StreamCallback | None = None,
+        context_window: int = 20,
+    ) -> SessionActorInvokeResult:
+        """Invoke the LLM actor bound to a session with a user prompt.
+
+        Resolves the actor's LLM provider via the provider registry, builds
+        conversation history from the session's existing messages, and invokes
+        the LLM.  Falls back to a stub response when no provider is configured.
+
+        Args:
+            session_id: The ULID of the session to invoke the actor for.
+            prompt: The user prompt to send to the actor.
+            actor_name: Override actor name (``provider/model`` format).
+                        Defaults to the session's bound actor.
+            stream_callback: Optional callable ``(chunk: str) -> None`` that
+                             receives streamed token chunks in real time.
+            context_window: Maximum number of historical messages to include
+                            in the LLM context.  Defaults to 20.
+
+        Returns:
+            A :class:`SessionActorInvokeResult` with the response, token
+            usage, and execution metadata.
+
+        Raises:
+            SessionNotFoundError: If no session with the given ID exists.
+            ValueError: If the actor's provider is configured but invalid
+                        (e.g. missing API key, unknown provider type).
+        """
+        session = self._session_repo.get_by_id(session_id)
+        if session is None:
+            raise SessionNotFoundError(f"Session '{session_id}' not found")
+
+        resolved_actor = actor_name or session.actor_name or "local/orchestrator"
+        history = session.get_messages()
+
+        start = time.monotonic()
+
+        if self._provider_registry is None:
+            # No provider configured — return stub response immediately.
+            return self._stub_response(resolved_actor, prompt, start, stream_callback)
+
+        # Resolve provider/model from actor name.
+        provider_type: str | None
+        model_id: str | None
+        if "/" in resolved_actor:
+            parts = resolved_actor.split("/", 1)
+            provider_type = parts[0] if parts[0] != "local" else None
+            model_id = parts[1] if parts[0] != "local" else None
+        else:
+            provider_type = None
+            model_id = resolved_actor
+
+        # create_llm raises ValueError for missing/invalid provider config.
+        # Cast to LlmHandle — all supported providers return a BaseChatModel
+        # which satisfies the Protocol at runtime.
+        llm: LlmHandle = self._provider_registry.create_llm(  # type: ignore[assignment]
+            provider_type=provider_type,
+            model_id=model_id,
+        )
+
+        messages = self._build_messages(history, prompt, context_window)
+        return self._invoke_llm(llm, messages, resolved_actor, start, stream_callback)
+
+    # ------------------------------------------------------------------
+    # Private helpers for invoke_actor
+    # ------------------------------------------------------------------
+
+    def _build_messages(
+        self,
+        history: list[SessionMessage],
+        prompt: str,
+        context_window: int,
+    ) -> list[Any]:
+        """Build the LangChain message list from session history and prompt."""
+        from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
+
+        msgs: list[Any] = [SystemMessage(content=self._ACTOR_SYSTEM_PROMPT)]
+        for msg in history[-context_window:]:
+            if msg.role == MessageRole.USER:
+                msgs.append(HumanMessage(content=msg.content))
+            elif msg.role == MessageRole.ASSISTANT:
+                msgs.append(AIMessage(content=msg.content))
+        msgs.append(HumanMessage(content=prompt))
+        return msgs
+
+    def _extract_token_usage(self, response: Any) -> tuple[int, int]:
+        """Extract (input_tokens, output_tokens) from an LLM response object."""
+        if hasattr(response, "usage_metadata") and response.usage_metadata:
+            input_tokens = int(getattr(response.usage_metadata, "input_tokens", 0) or 0)
+            output_tokens = int(
+                getattr(response.usage_metadata, "output_tokens", 0) or 0
+            )
+            return input_tokens, output_tokens
+        if hasattr(response, "response_metadata"):
+            meta: dict[str, Any] = response.response_metadata or {}
+            token_usage: dict[str, Any] = (
+                meta.get("token_usage") or meta.get("usage") or {}
+            )
+            input_tokens = int(
+                token_usage.get("prompt_tokens") or token_usage.get("input_tokens") or 0
+            )
+            output_tokens = int(
+                token_usage.get("completion_tokens")
+                or token_usage.get("output_tokens")
+                or 0
+            )
+            return input_tokens, output_tokens
+        return 0, 0
+
+    def _invoke_llm(
+        self,
+        llm: LlmHandle,
+        messages: list[Any],
+        resolved_actor: str,
+        start: float,
+        stream_callback: StreamCallback | None,
+    ) -> SessionActorInvokeResult:
+        """Invoke the LLM (streaming or non-streaming) and return a result."""
+        response_content = ""
+        input_tokens = 0
+        output_tokens = 0
+
+        if stream_callback is not None:
+            for chunk in llm.stream(messages):
+                raw = chunk.content
+                chunk_text = raw if isinstance(raw, str) else str(raw)
+                response_content += chunk_text
+                stream_callback(chunk_text)
+        else:
+            response = llm.invoke(messages)
+            raw_content = response.content
+            response_content = (
+                raw_content if isinstance(raw_content, str) else str(raw_content)
+            )
+            input_tokens, output_tokens = self._extract_token_usage(response)
+
+        duration_s = time.monotonic() - start
+        total_tokens = input_tokens + output_tokens
+        _estimate_cost = getattr(self._provider_registry, "estimate_cost", None)
+        cost_usd: float = (
+            _estimate_cost(resolved_actor, input_tokens, output_tokens)
+            if _estimate_cost is not None
+            else total_tokens * 0.000002
+        )
+        cost_str = f"${cost_usd:.4f}"
+
+        return SessionActorInvokeResult(
+            response=response_content,
+            commands_executed=[],
+            result={},
+            input_tokens=input_tokens,
+            output_tokens=output_tokens,
+            cost=cost_str,
+            duration_s=round(duration_s, 1),
+            tool_calls=0,
+            actor=resolved_actor,
+        )
+
+    def _stub_response(
+        self,
+        resolved_actor: str,
+        prompt: str,
+        start: float,
+        stream_callback: StreamCallback | None,
+    ) -> SessionActorInvokeResult:
+        """Return a stub response when no provider is configured."""
+        stub_content = f"Acknowledged: {prompt[:100]}"
+        if stream_callback is not None:
+            stream_callback(stub_content)
+        duration_s = time.monotonic() - start
+        return SessionActorInvokeResult(
+            response=stub_content,
+            commands_executed=[],
+            result={},
+            input_tokens=0,
+            output_tokens=0,
+            cost="$0.0000",
+            duration_s=round(duration_s, 1),
+            tool_calls=0,
+            actor=resolved_actor,
+        )

src/cleveragents/cli/commands/session.py

@@ -27,11 +27,20 @@ from rich.panel import Panel
 from rich.table import Table
 
 from cleveragents.a2a.models import A2aRequest
+from cleveragents.cli.commands.session_tell_panels import (
+    make_stream_callback,
+    render_streaming_session_panel,
+    render_streaming_structured,
+    render_streaming_usage_panel,
+    render_tell_rich,
+    render_tell_structured,
+)
 from cleveragents.cli.formatting import OutputFormat, format_output
 from cleveragents.core.exceptions import DatabaseError
 from cleveragents.domain.models.core.session import (
     MessageRole,
     Session,
+    SessionActorInvokeResult,
     SessionExportError,
     SessionImportError,
     SessionMessage,
@@ -697,50 +706,114 @@ def tell(
         bool,
         typer.Option("--stream", help="Stream response in real-time"),
     ] = False,
+    fmt: Annotated[
+        str,
+        typer.Option("--format", "-f", help=_FORMAT_HELP),
+    ] = "rich",
 ) -> None:
     """Send a message to a session.
 
-    Appends a user message and generates an assistant response. For M3, the
-    actor execution is stubbed — the assistant echoes an acknowledgement.
+    Sends a natural-language request to the orchestrator actor bound to the
+    session. The orchestrator interprets the request and issues the necessary
+    CleverAgents commands. Use ``--stream`` to see responses in real time.
 
     Examples:
         agents session tell --session 01HXYZ... "Hello, world"
         agents session tell --session 01HXYZ... --actor openai/gpt-4 "Plan a feature"
         agents session tell --session 01HXYZ... --stream "Build tests"
+        agents session tell --session 01HXYZ... --format json "What can you do?"
     """
     try:
         service = _get_session_service()
 
-        # Append user message
+        # Resolve actor name from session before appending the user message.
+        session = service.get(session_id)
+        resolved_actor = actor or session.actor_name or "local/orchestrator"
+
+        # Persist the user message so the service layer includes it in history.
         service.append_message(
             session_id=session_id,
             role=MessageRole.USER,
             content=prompt,
         )
 
-        # Stub actor execution: generate simple assistant response
-        assistant_content = (
-            f"Acknowledged: {prompt[:100]}"
-            if not actor
-            else f"[{actor}] Acknowledged: {prompt[:100]}"
-        )
-        service.append_message(
-            session_id=session_id,
-            role=MessageRole.ASSISTANT,
-            content=assistant_content,
-        )
-
         if stream:
-            # Simulate streaming by printing character by character
-            for char in assistant_content:
-                sys.stdout.write(char)
-                sys.stdout.flush()
-            sys.stdout.write("\n")
-        else:
-            from rich.markup import escape
+            # ── Streaming mode ──────────────────────────────────────────────
+            if fmt not in (OutputFormat.RICH.value, OutputFormat.COLOR.value):
+                # Non-rich streaming: invoke without a callback, then format.
+                result: SessionActorInvokeResult = service.invoke_actor(
+                    session_id=session_id,
+                    prompt=prompt,
+                    actor_name=resolved_actor,
+                )
+                service.append_message(
+                    session_id=session_id,
+                    role=MessageRole.ASSISTANT,
+                    content=result.response,
+                )
+                service.update_token_usage(
+                    session_id=session_id,
+                    input_tokens=result.input_tokens,
+                    output_tokens=result.output_tokens,
+                    cost=float(result.cost.lstrip("$")),
+                )
+                render_streaming_structured(result, session_id, prompt, fmt)
+                return
 
-            console.print(f"[dim]user:[/dim] {escape(prompt)}")
-            console.print(f"[cyan]assistant:[/cyan] {escape(assistant_content)}")
+            # Rich streaming: Session panel → stream tokens → Usage panel.
+            render_streaming_session_panel(session_id, resolved_actor)
+
+            streamed_tokens: list[str] = []
+            stream_cb = make_stream_callback(streamed_tokens)
+
+            result = service.invoke_actor(
+                session_id=session_id,
+                prompt=prompt,
+                actor_name=resolved_actor,
+                stream_callback=stream_cb,
+            )
+            sys.stdout.write("\n")
+            sys.stdout.flush()
+
+            full_response = "".join(streamed_tokens) or result.response
+            service.append_message(
+                session_id=session_id,
+                role=MessageRole.ASSISTANT,
+                content=full_response,
+            )
+            service.update_token_usage(
+                session_id=session_id,
+                input_tokens=result.input_tokens,
+                output_tokens=result.output_tokens,
+                cost=float(result.cost.lstrip("$")),
+            )
+            render_streaming_usage_panel(result)
+
+        else:
+            # ── Non-streaming mode ──────────────────────────────────────────
+            result = service.invoke_actor(
+                session_id=session_id,
+                prompt=prompt,
+                actor_name=resolved_actor,
+            )
+
+            service.append_message(
+                session_id=session_id,
+                role=MessageRole.ASSISTANT,
+                content=result.response,
+            )
+            service.update_token_usage(
+                session_id=session_id,
+                input_tokens=result.input_tokens,
+                output_tokens=result.output_tokens,
+                cost=float(result.cost.lstrip("$")),
+            )
+
+            if fmt not in (OutputFormat.RICH.value, OutputFormat.COLOR.value):
+                render_tell_structured(result, session_id, prompt, "review", fmt)
+                return
+
+            render_tell_rich(result, session_id, prompt, "review")
 
     except SessionNotFoundError as exc:
         console.print(f"[red]Session not found:[/red] {session_id}")

src/cleveragents/cli/commands/session_tell_panels.py

@@ -0,0 +1,274 @@
+"""Presentation helpers for the ``agents session tell`` command.
+
+Renders the spec-required Rich output panels and structured (JSON/YAML/plain)
+output envelopes for both streaming and non-streaming tell responses.
+
+Extracted from ``session.py`` to keep that module within the 500-line limit
+mandated by CONTRIBUTING.md.
+
+Spec reference: ``docs/specification.md`` section "agents session tell"
+(lines 2234-2390 for non-streaming, 2397-2510 for streaming).
+"""
+
+from __future__ import annotations
+
+import sys
+from collections.abc import Callable
+from typing import Any
+
+from rich.console import Console
+from rich.markup import escape
+from rich.panel import Panel
+
+from cleveragents.cli.formatting import format_output
+from cleveragents.domain.models.core.session import SessionActorInvokeResult
+
+_console = Console()
+
+
+# ---------------------------------------------------------------------------
+# Non-streaming output
+# ---------------------------------------------------------------------------
+
+
+def render_tell_rich(
+    result: SessionActorInvokeResult,
+    session_id: str,
+    prompt: str,
+    automation: str,
+) -> None:
+    """Render the four spec-required Rich panels for non-streaming tell.
+
+    Panels rendered (in order):
+    1. **Plan Request** — Actor, Session, Automation, Prompt
+    2. **Commands Executed** — list of commands or empty-state message
+    3. **Result** — created/modified artifacts or empty-state message
+    4. **Usage** — Input Tokens, Output Tokens, Cost, Duration, Tool Calls
+
+    Followed by the spec success message:
+    ``✓ OK Orchestrator completed N commands``
+
+    Args:
+        result: The actor invocation result.
+        session_id: The session ULID.
+        prompt: The original user prompt.
+        automation: The automation profile name (from session/action config).
+    """
+    prompt_preview = prompt[:50] + "..." if len(prompt) > 50 else prompt
+
+    # 1. Plan Request panel
+    plan_request_details = (
+        f"[magenta bold]Actor:[/magenta bold] {escape(result.actor)}\n"
+        f"[cyan bold]Session:[/cyan bold] {escape(session_id)}\n"
+        f"[magenta bold]Automation:[/magenta bold] {escape(automation)}\n"
+        f"[blue]Prompt:[/blue] {escape(prompt_preview)}"
+    )
+    _console.print(Panel(plan_request_details, title="Plan Request", expand=False))
+
+    # 2. Commands Executed panel
+    if result.commands_executed:
+        cmds_text = "\n".join(f"- {escape(cmd)}" for cmd in result.commands_executed)
+    else:
+        cmds_text = "[dim](no commands executed)[/dim]"
+    _console.print(Panel(cmds_text, title="Commands Executed", expand=False))
+
+    # 3. Result panel
+    if result.result:
+        result_text = "\n".join(
+            f"[green bold]{escape(k)}:[/green bold] {escape(str(v))}"
+            for k, v in result.result.items()
+        )
+    else:
+        result_text = "[dim](no structured result)[/dim]"
+    _console.print(Panel(result_text, title="Result", expand=False))
+
+    # 4. Usage panel
+    usage_text = (
+        f"[blue]Input Tokens:[/blue] {result.input_tokens:,}\n"
+        f"[blue]Output Tokens:[/blue] {result.output_tokens:,}\n"
+        f"[yellow bold]Cost:[/yellow bold] {escape(result.cost)}\n"
+        f"[green]Duration:[/green] {result.duration_s}s\n"
+        f"[blue]Tool Calls:[/blue] {result.tool_calls}"
+    )
+    _console.print(Panel(usage_text, title="Usage", expand=False))
+
+    num_commands = len(result.commands_executed)
+    _console.print(
+        f"[green]✓ OK[/green] Orchestrator completed {num_commands} commands"
+    )
+
+
+def render_tell_structured(
+    result: SessionActorInvokeResult,
+    session_id: str,
+    prompt: str,
+    automation: str,
+    fmt: str,
+) -> None:
+    """Render structured (JSON/YAML/plain) output for non-streaming tell.
+
+    Wraps the data in the full spec-defined envelope::
+
+        {
+          "command": "agents session tell ...",
+          "status": "ok",
+          "exit_code": 0,
+          "data": { "plan_request": {...}, "commands_executed": [...], ... },
+          "timing": { "duration_ms": ... },
+          "messages": [{ "level": "ok", "text": "Orchestrator completed N commands" }]
+        }
+
+    Args:
+        result: The actor invocation result.
+        session_id: The session ULID.
+        prompt: The original user prompt.
+        automation: The automation profile name.
+        fmt: Output format string (``"json"``, ``"yaml"``, ``"plain"``).
+    """
+    num_commands = len(result.commands_executed)
+    duration_ms = int(result.duration_s * 1000)
+    prompt_preview = prompt[:100] + ("..." if len(prompt) > 100 else "")
+
+    envelope: dict[str, Any] = {
+        "command": (f'agents session tell "{prompt_preview}" --session {session_id}'),
+        "status": "ok",
+        "exit_code": 0,
+        "data": {
+            "plan_request": {
+                "actor": result.actor,
+                "session": session_id,
+                "automation": automation,
+                "prompt": prompt_preview,
+            },
+            "commands_executed": result.commands_executed,
+            "result": result.result,
+            "usage": {
+                "input_tokens": result.input_tokens,
+                "output_tokens": result.output_tokens,
+                "cost": result.cost,
+                "duration_s": result.duration_s,
+                "tool_calls": result.tool_calls,
+            },
+        },
+        "timing": {"duration_ms": duration_ms},
+        "messages": [
+            {
+                "level": "ok",
+                "text": f"Orchestrator completed {num_commands} commands",
+            }
+        ],
+    }
+    format_output(envelope, fmt)
+
+
+# ---------------------------------------------------------------------------
+# Streaming output
+# ---------------------------------------------------------------------------
+
+
+def render_streaming_session_panel(
+    session_id: str,
+    actor: str,
+) -> None:
+    """Render the spec-required Session panel before streaming begins.
+
+    Args:
+        session_id: The session ULID.
+        actor: The resolved actor name.
+    """
+    session_details = (
+        f"[cyan bold]ID:[/cyan bold] {escape(session_id)}\n"
+        f"[blue]Actor:[/blue] {escape(actor)}\n"
+        f"[blue]Mode:[/blue] streaming"
+    )
+    _console.print(Panel(session_details, title="Session", expand=False))
+
+
+def render_streaming_usage_panel(result: SessionActorInvokeResult) -> None:
+    """Render the spec-required Usage panel after streaming completes.
+
+    Args:
+        result: The actor invocation result (with token/timing data).
+    """
+    total_tokens = result.input_tokens + result.output_tokens
+    usage_details = (
+        f"[blue]Tokens:[/blue] {total_tokens:,} (stream)\n"
+        f"[blue]Duration:[/blue] {result.duration_s}s\n"
+        f"[blue]Tool Calls:[/blue] {result.tool_calls}"
+    )
+    _console.print(Panel(usage_details, title="Usage", expand=False))
+    _console.print("[green]✓ OK[/green] Stream complete")
+
+
+def render_streaming_structured(
+    result: SessionActorInvokeResult,
+    session_id: str,
+    prompt: str,
+    fmt: str,
+) -> None:
+    """Render structured (JSON/YAML/plain) output for streaming tell.
+
+    Wraps the data in the full spec-defined streaming envelope::
+
+        {
+          "command": "agents session tell --session ... --stream ...",
+          "status": "ok",
+          "exit_code": 0,
+          "data": { "session": {...}, "response": "...", "usage": {...} },
+          "timing": { "duration_ms": ... },
+          "messages": [{ "level": "ok", "text": "Stream complete" }]
+        }
+
+    Args:
+        result: The actor invocation result.
+        session_id: The session ULID.
+        prompt: The original user prompt.
+        fmt: Output format string (``"json"``, ``"yaml"``, ``"plain"``).
+    """
+    duration_ms = int(result.duration_s * 1000)
+    total_tokens = result.input_tokens + result.output_tokens
+    prompt_preview = prompt[:100] + ("..." if len(prompt) > 100 else "")
+
+    envelope: dict[str, Any] = {
+        "command": (
+            f'agents session tell --session {session_id} --stream "{prompt_preview}"'
+        ),
+        "status": "ok",
+        "exit_code": 0,
+        "data": {
+            "session": {
+                "id": session_id,
+                "actor": result.actor,
+                "mode": "streaming",
+            },
+            "response": result.response,
+            "usage": {
+                "tokens": total_tokens,
+                "duration_s": result.duration_s,
+                "tool_calls": result.tool_calls,
+            },
+        },
+        "timing": {"duration_ms": duration_ms},
+        "messages": [{"level": "ok", "text": "Stream complete"}],
+    }
+    format_output(envelope, fmt)
+
+
+def make_stream_callback(
+    streamed_tokens: list[str],
+) -> Callable[[str], None]:
+    """Create a streaming callback that writes chunks to stdout.
+
+    Args:
+        streamed_tokens: Mutable list that accumulates streamed chunks.
+
+    Returns:
+        A callable ``(chunk: str) -> None``.
+    """
+
+    def _cb(chunk: str) -> None:
+        sys.stdout.write(chunk)
+        sys.stdout.flush()
+        streamed_tokens.append(chunk)
+
+    return _cb

src/cleveragents/domain/models/core/__init__.py

@@ -279,6 +279,7 @@ from cleveragents.domain.models.core.sandbox_strategy import (
 from cleveragents.domain.models.core.session import (
     MessageRole,
     Session,
+    SessionActorInvokeResult,
     SessionExportError,
     SessionImportError,
     SessionMessage,
@@ -504,6 +505,7 @@ __all__ = [
     "ServiceRetryPolicy",
     "ServiceRetryPolicyRegistry",
     "Session",
+    "SessionActorInvokeResult",
     "SessionCostBudget",
     "SessionExportError",
     "SessionImportError",

src/cleveragents/domain/models/core/session.py

@@ -38,6 +38,7 @@ import json
 import re
 from abc import ABC, abstractmethod
 from collections import OrderedDict
+from dataclasses import dataclass, field
 from datetime import datetime
 from enum import StrEnum
 from typing import Any
@@ -479,6 +480,42 @@ class Session(BaseModel):
 # ---------------------------------------------------------------------------
 
 
+@dataclass
+class SessionActorInvokeResult:
+    """Result of invoking an LLM actor for a session tell operation.
+
+    Returned by :meth:`SessionService.invoke_actor` and consumed by the CLI
+    presentation layer to render the spec-required output panels.
+    """
+
+    response: str
+    """Full LLM response text."""
+
+    commands_executed: list[str] = field(default_factory=list)
+    """CLI commands the actor executed (may be empty for stub responses)."""
+
+    result: dict[str, Any] = field(default_factory=dict)
+    """Structured result data (created/modified artifacts, etc.)."""
+
+    input_tokens: int = 0
+    """Number of input tokens consumed."""
+
+    output_tokens: int = 0
+    """Number of output tokens generated."""
+
+    cost: str = "$0.0000"
+    """Estimated cost string, e.g. ``"$0.0023"``."""
+
+    duration_s: float = 0.0
+    """Wall-clock duration in seconds."""
+
+    tool_calls: int = 0
+    """Number of tool/function calls made by the actor."""
+
+    actor: str = "local/orchestrator"
+    """Resolved actor name used for the invocation."""
+
+
 class SessionServiceError(Exception):
     """Base error for session service operations."""
 
@@ -517,7 +554,7 @@ class SessionService(ABC):
     | ``agents session delete`` | ``delete()``          |
     | ``agents session export`` | ``export_session()``  |
     | ``agents session import`` | ``import_session()``  |
-    | ``agents session tell``   | ``append_message()``  |
+    | ``agents session tell``   | ``invoke_actor()``    |
     """
 
     @abstractmethod
@@ -635,3 +672,41 @@ class SessionService(ABC):
         Raises:
             SessionNotFoundError: If no session with the given ID exists.
         """
+
+    @abstractmethod
+    def invoke_actor(
+        self,
+        session_id: str,
+        prompt: str,
+        actor_name: str | None = None,
+        stream_callback: Any | None = None,
+        context_window: int = 20,
+    ) -> SessionActorInvokeResult:
+        """Invoke the LLM actor bound to a session with a user prompt.
+
+        Resolves the actor's LLM provider via the provider registry, builds
+        conversation history from the session's existing messages, and invokes
+        the LLM.  Falls back gracefully when no provider is configured.
+
+        The caller is responsible for persisting the user message (via
+        :meth:`append_message`) *before* calling this method so that the
+        history passed to the LLM includes the current turn.
+
+        Args:
+            session_id: The ULID of the session to invoke the actor for.
+            prompt: The user prompt to send to the actor.
+            actor_name: Override actor name (``provider/model`` format).
+                        Defaults to the session's bound actor.
+            stream_callback: Optional callable ``(chunk: str) -> None`` that
+                             receives streamed token chunks in real time.
+            context_window: Maximum number of historical messages to include
+                            in the LLM context.  Defaults to 20.
+
+        Returns:
+            A :class:`SessionActorInvokeResult` with the response, token
+            usage, and execution metadata.
+
+        Raises:
+            SessionNotFoundError: If no session with the given ID exists.
+            ValueError: If the actor's provider is configured but invalid.
+        """

src/cleveragents/providers/__init__.py

@@ -7,6 +7,7 @@ fallback selection, and provider registry management.
 from .cost_table import CostEntry, ProviderCostTable
 from .cost_tracker import BudgetCheckResult, BudgetStatus, CostTracker
 from .fallback_selector import FallbackResult, FallbackSelector
+from .llm_protocol import LlmChunk, LlmHandle, LlmResponse, StreamCallback
 from .registry import (
     ProviderRegistry,
     get_provider_registry,
@@ -21,8 +22,12 @@ __all__ = [
     "CostTracker",
     "FallbackResult",
     "FallbackSelector",
+    "LlmChunk",
+    "LlmHandle",
+    "LlmResponse",
     "ProviderCostTable",
     "ProviderRegistry",
+    "StreamCallback",
     "get_provider_registry",
     "reset_provider_registry",
     "resolve_provider_by_name",

src/cleveragents/providers/llm_protocol.py

@@ -0,0 +1,93 @@
+"""Protocol interface for LLM handles used across the application layer.
+
+Defines :class:`LlmHandle` — a structural Protocol that captures the minimal
+interface required to invoke a language model (both streaming and
+non-streaming).  All supported LangChain chat-model classes satisfy this
+Protocol at runtime; the Protocol is defined here so that the type-checker
+can verify call sites without requiring ``langchain_core`` to be installed in
+the type-checking virtual environment.
+
+Usage::
+
+    from cleveragents.providers.llm_protocol import LlmHandle, LlmResponse
+
+    def call_llm(llm: LlmHandle, messages: list[object]) -> str:
+        response: LlmResponse = llm.invoke(messages)
+        content = response.content
+        return content if isinstance(content, str) else str(content)
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Iterator
+from typing import Any, Protocol, runtime_checkable
+
+
+class LlmResponse(Protocol):
+    """Minimal interface for a non-streaming LLM response message.
+
+    Satisfied by ``langchain_core.messages.AIMessage`` and any other
+    message type that exposes a ``content`` attribute.
+    """
+
+    @property
+    def content(self) -> str | list[str | dict[str, Any]]:
+        """The text content of the response."""
+        ...
+
+    @property
+    def usage_metadata(self) -> dict[str, Any] | None:
+        """Optional token-usage metadata dict (may be ``None``)."""
+        ...
+
+    @property
+    def response_metadata(self) -> dict[str, Any]:
+        """Provider-specific response metadata (always a dict)."""
+        ...
+
+
+class LlmChunk(Protocol):
+    """Minimal interface for a streaming LLM response chunk.
+
+    Satisfied by ``langchain_core.messages.AIMessageChunk``.
+    """
+
+    @property
+    def content(self) -> str | list[str | dict[str, Any]]:
+        """The text content of this chunk."""
+        ...
+
+
+@runtime_checkable
+class LlmHandle(Protocol):
+    """Structural Protocol for a LangChain-compatible chat model handle.
+
+    Any object that implements ``invoke`` and ``stream`` with compatible
+    signatures satisfies this Protocol.  In production this will always be
+    a ``langchain_core.language_models.BaseChatModel`` subclass returned by
+    :meth:`~cleveragents.providers.registry.ProviderRegistry.create_llm`.
+
+    The Protocol is intentionally narrow — it only captures the two methods
+    used by the application layer — so that mock objects in tests can satisfy
+    it without inheriting from LangChain base classes.
+    """
+
+    def invoke(
+        self,
+        input: list[Any],
+        **kwargs: Any,
+    ) -> LlmResponse:
+        """Invoke the model and return a single response."""
+        ...
+
+    def stream(
+        self,
+        input: list[Any],
+        **kwargs: Any,
+    ) -> Iterator[LlmChunk]:
+        """Stream the model response as an iterator of chunks."""
+        ...
+
+
+# Convenience type alias for stream callbacks used in the CLI layer.
+StreamCallback = Callable[[str], None]