fix(cli): fix session tell output to return spec-required structured response #6729

Open
HAL9000 wants to merge 5 commits from fix/issue-6452-session-tell-output into master
9 changed files with 817 additions and 444 deletions
+9
View File
@@ -5,6 +5,15 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
## [Unreleased]
### Fixed
- **Session Tell Structured Output** (#6452): `agents session tell` now renders
spec-required Rich panels (Plan Request, Commands Executed, Result, Usage)
instead of a plain echo. Adds `--format json/yaml/plain/table/color` option
with a spec-compliant JSON envelope. Tell helpers extracted into
`session_tell.py` and `session_helpers.py` to keep `session.py` under the
500-line project limit.
### Added
- **Git Worktree Sandbox Apply** (#4454): The `plan apply` command now merges
+2
View File
@@ -3,6 +3,7 @@
* Aditya Chhabra <aditya.chhabra@cleverthis.com>
* Brent E. Edwards <brent.edwards@cleverthis.com>
* Hamza Khyari <hamza.khyari@cleverthis.com>
* HAL9000 <hal9000@cleverthis.com>
* Jeffrey Phillips Freeman <jeffrey.freeman@syncleus.com>
* Luis Mendes <luis.p.mendes@gmail.com>
* Rui Hu <rui.hu@cleverthis.com>
@@ -13,4 +14,5 @@ Below are some of the specific details of various contributions.
* Jeffrey Phillips Freeman has acted as Lead Developer, daily contributor, and Project Owner.
* Brent E. Edwards has contributed quality assurance, test coverage, and CI pipeline improvements.
* HAL9000 has implemented session tell structured output (#6452), contributing spec-compliant CLI panels (Plan Request, Commands Executed, Result, Usage) with multi-format support.
* This project was made possible thanks to considerable donation of time, money, and resources by CleverThis, Inc.
+18 -3
View File
@@ -168,18 +168,33 @@ Feature: Session CLI commands
And the session CLI output should contain "Invalid JSON"
# Tell command tests
Scenario: Tell appends message to session
@tdd_issue @tdd_issue_6452
Scenario: Tell renders structured output panels
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"
And the session CLI output should contain "Commands Executed"
And the session CLI output should contain "Result"
And the session CLI output should contain "Usage"
And the session CLI output should contain "Orchestrator acknowledged request"
Scenario: Tell with custom actor
@tdd_issue @tdd_issue_6452
Scenario: Tell with custom actor displays actor in 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"
@tdd_issue @tdd_issue_6452
Scenario: Tell with JSON format returns structured envelope
Given there is a mocked session for tell
When I run session CLI tell with --format json and prompt "Review session"
Then the session CLI tell should succeed
And the session CLI output should be valid JSON
And the session CLI JSON should contain "plan_request"
And the session CLI JSON should contain "usage"
Scenario: Tell to non-existent session
When I run session CLI tell to a non-existent session
Then the session CLI should exit with error
+14 -2
View File
@@ -460,10 +460,22 @@ def step_tell_with_actor(context: Context, actor: str, prompt: str) -> None:
)
@when('I run session CLI tell with --format json and prompt "{prompt}"')
def step_tell_with_json_format(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.result = context.runner.invoke(
session_app,
["tell", "--session", context.session_id, "--format", "json", prompt],
)
@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 'NONEXISTENT' not found"
)
context.result = context.runner.invoke(
session_app,
+49 -1
View File
@@ -292,12 +292,59 @@ def tell_message() -> 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"Missing 'Plan Request' panel:\n{result.output}"
)
assert "Commands Executed" in result.output, (
f"Missing 'Commands Executed' panel:\n{result.output}"
)
assert "Result" in result.output, (
f"Missing 'Result' panel:\n{result.output}"
)
assert "Usage" in result.output, (
f"Missing 'Usage' panel:\n{result.output}"
)
assert "Orchestrator acknowledged request" in result.output, (
f"Missing status text:\n{result.output}"
)
print("session-cli-tell-message-ok")
finally:
_teardown()
def tell_json_envelope() -> None:
"""Test that ``session tell --format json`` returns a spec-compliant envelope."""
import json as _json
sid = str(ULID())
svc = _setup_service()
svc.get.return_value = _mock_session(session_id=sid)
svc.append_message.side_effect = [
_mock_message(MessageRole.USER, "Review session", 0),
_mock_message(MessageRole.ASSISTANT, "Acknowledged: Review session", 1),
]
try:
result = runner.invoke(
session_app,
["tell", "--session", sid, "--format", "json", "Review session"],
)
assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
parsed = _json.loads(result.output)
# Verify top-level envelope keys
assert "status" in parsed, f"Missing 'status' in envelope: {parsed}"
assert "data" in parsed, f"Missing 'data' in envelope: {parsed}"
data = parsed["data"]
assert "plan_request" in data, f"Missing 'plan_request' in data: {data}"
assert "usage" in data, f"Missing 'usage' in data: {data}"
assert "commands_executed" in data, (
f"Missing 'commands_executed' in data: {data}"
)
assert "result" in data, f"Missing 'result' in data: {data}"
print("session-cli-tell-json-envelope-ok")
finally:
_teardown()
def export_rich_panels() -> None:
"""Test that export renders all three spec-required Rich panels."""
sid = str(ULID())
@@ -379,6 +426,7 @@ _COMMANDS: dict[str, object] = {
"export-rich-panels": export_rich_panels,
"export-stdout-rich-panels": export_stdout_rich_panels,
"tell-message": tell_message,
"tell-json-envelope": tell_json_envelope,
}
if __name__ == "__main__":
+12 -2
View File
@@ -82,8 +82,18 @@ Session Export Stdout Rich Panels
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} session-cli-export-stdout-rich-panels-ok
Session Tell Appends Message
[Documentation] Verify that ``session tell`` appends a message
Session Tell Renders Structured Panels
[Documentation] Verify that ``session tell`` renders spec-required Rich panels
${result}= Run Process ${PYTHON} ${HELPER} tell-message cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} session-cli-tell-message-ok
Session Tell JSON Envelope
[Documentation] Verify that ``session tell --format json`` returns a structured envelope
${result}= Run Process ${PYTHON} ${HELPER} tell-json-envelope cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} session-cli-tell-json-envelope-ok
+69 -436
View File
@@ -14,27 +14,39 @@ task A7.cli.
from __future__ import annotations
import contextlib
import io
import json
import logging
import sys
from collections import OrderedDict
from pathlib import Path
from typing import Annotated, Any, cast
import typer
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from cleveragents.a2a.models import A2aRequest
from cleveragents.cli.commands.session_helpers import (
build_export_content,
render_create_rich,
render_delete_rich,
render_export_panels,
render_import_rich,
render_list_rich,
render_show_rich,
session_list_dict,
session_summary_dict,
)
from cleveragents.cli.commands.session_tell import (
build_tell_command,
build_tell_payload,
render_tell_output,
)
from cleveragents.cli.formatting import OutputFormat, format_output
from cleveragents.core.exceptions import DatabaseError
from cleveragents.domain.models.core.session import (
MessageRole,
Session,
SessionExportError,
SessionImportError,
SessionMessage,
SessionNotFoundError,
SessionService,
)
@@ -48,18 +60,11 @@ _log = logging.getLogger(__name__)
# Reusable --format option description
_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
# ---------------------------------------------------------------------------
# Module-level service accessor (patchable in tests)
# ---------------------------------------------------------------------------
_service: SessionService | None = None
def _get_session_service() -> SessionService:
"""Get or create the SessionService instance.
Production usage goes through the DI container's ``session_service``
provider; tests can patch ``_service`` or this function directly.
"""
"""Get or create the SessionService instance."""
global _service
if _service is not None:
return _service
@@ -79,18 +84,7 @@ def _reset_session_service() -> None:
def _facade_dispatch(operation: str, params: dict[str, Any]) -> dict[str, Any]:
"""Route an operation through the A2A local facade.
Returns the response data dict on success. Raises on error
so that CLI commands can render the appropriate error message.
stdout/stderr are redirected during facade construction to prevent
structlog or Rich output from polluting CLI output captured by
test runners.
"""
import contextlib
import io
"""Route an operation through the A2A local facade."""
with (
contextlib.redirect_stdout(io.StringIO()),
contextlib.redirect_stderr(io.StringIO()),
@@ -105,68 +99,6 @@ def _facade_dispatch(operation: str, params: dict[str, Any]) -> dict[str, Any]:
return dict(response.result or {})
# ---------------------------------------------------------------------------
# Helpers — dict builders and rich printing
# ---------------------------------------------------------------------------
def _session_summary_dict(session: Session) -> OrderedDict[str, Any]:
"""Build a stable-ordered summary dict for a session."""
result: OrderedDict[str, Any] = OrderedDict()
result["session_id"] = session.session_id
result["actor"] = session.actor_name or "(none)"
result["namespace"] = session.namespace
result["messages"] = session.message_count
result["created"] = session.created_at.isoformat()
result["updated"] = session.updated_at.isoformat()
return result
def _session_list_dict(sessions: list[Session]) -> dict[str, Any]:
"""Build the list output with summary stats per spec."""
items = []
for s in sessions:
items.append(
{
"id": s.session_id,
"name": s.name or None,
"actor": s.actor_name or "(none)",
"messages": s.message_count,
"updated": s.updated_at.isoformat(),
}
)
# Build summary section per spec
total_messages = sum(s.message_count for s in sessions)
# Find most recent and oldest sessions
if sessions:
sorted_sessions = sorted(sessions, key=lambda x: x.updated_at, reverse=True)
most_recent = sorted_sessions[0].name or sorted_sessions[0].session_id[:8]
oldest = sorted_sessions[-1].name or sorted_sessions[-1].session_id[:8]
else:
most_recent = None
oldest = None
summary = {
"total": len(sessions),
"most_recent": most_recent,
"oldest": oldest,
"total_messages": total_messages,
"storage": "0 KB", # Placeholder - actual storage calculation not implemented
}
return {
"sessions": items,
"summary": summary,
}
# ---------------------------------------------------------------------------
# CLI Commands
# ---------------------------------------------------------------------------
@app.command()
def create(
actor: Annotated[
@@ -180,74 +112,28 @@ def create(
) -> None:
"""Create a new interactive session.
Optionally bind the session to an orchestrator actor via ``--actor``.
Examples:
agents session create
agents session create --actor openai/gpt-4
agents session create --format json
"""
try:
# Create session through the service, then notify the A2A
# facade for protocol compliance and telemetry.
service = _get_session_service()
session = service.create(actor_name=actor)
# Notify the facade layer for A2A protocol bookkeeping.
# Pass session_id so the facade handler acknowledges the already-
# persisted session instead of creating a duplicate (#1141).
import contextlib
with contextlib.suppress(Exception):
_facade_dispatch(
"session.create",
{"actor_name": actor or "", "session_id": session.session_id},
)
data = _session_summary_dict(session)
data = session_summary_dict(session)
if fmt not in (OutputFormat.RICH.value, OutputFormat.COLOR.value):
typer.echo(format_output(dict(data), fmt))
return
details = (
f"[bold]Session ID:[/bold] {session.session_id}\n"
f"[bold]Actor:[/bold] {session.actor_name or '(none)'}\n"
f"[bold]Namespace:[/bold] {session.namespace}\n"
f"[bold]Created:[/bold] {session.created_at.strftime('%Y-%m-%d %H:%M')}"
)
console.print(Panel(details, title="Session", expand=False))
# Settings panel
settings_text = (
"[yellow]Automation:[/yellow] default\n"
"[yellow]Streaming:[/yellow] off\n"
"[yellow]Context:[/yellow] default\n"
"[yellow]Memory:[/yellow] enabled\n"
"[yellow]Max History:[/yellow] 50 turns"
)
console.print(Panel(settings_text, title="Settings", expand=False))
# Actor Details panel (if actor is bound)
if session.actor_name:
try:
from cleveragents.application.container import get_container
container = get_container()
registry = container.actor_registry()
actor_obj = registry.get_actor(session.actor_name)
actor_details = (
f"[blue]Provider:[/blue] {actor_obj.provider}\n"
f"[blue]Model:[/blue] {actor_obj.model}\n"
f"[blue]Temperature:[/blue] "
f"{getattr(actor_obj, 'temperature', 0.7)}\n"
"[blue]Context Window:[/blue] 200K tokens"
)
console.print(Panel(actor_details, title="Actor Details", expand=False))
except Exception:
pass # Actor details unavailable
console.print("[green]✓ OK[/green] Session created")
render_create_rich(session)
except SessionNotFoundError as exc:
console.print(f"[red]Error:[/red] {exc}")
@@ -270,8 +156,6 @@ def list_sessions(
) -> None:
"""List all sessions.
Displays session IDs, bound actors, message counts, and last update times.
Examples:
agents session list
agents session list --format json
@@ -289,8 +173,6 @@ def list_sessions(
raise typer.Exit(1) from exc
if not sessions:
# For machine-readable formats, always emit a structured empty list
# so that callers parsing JSON/YAML receive valid output.
if fmt not in (OutputFormat.RICH.value, OutputFormat.COLOR.value):
typer.echo(format_output({"sessions": [], "total": 0}, fmt))
return
@@ -298,48 +180,13 @@ def list_sessions(
console.print("Create one with 'agents session create'")
return
data = _session_list_dict(sessions)
data = session_list_dict(sessions)
if fmt not in (OutputFormat.RICH.value, OutputFormat.COLOR.value):
typer.echo(format_output(data, fmt))
return
# Rich table
table = Table(title="Sessions", show_header=True, border_style="blue")
table.add_column("ID", style="cyan")
table.add_column("Name", style="magenta")
table.add_column("Actor", style="blue")
table.add_column("Messages", justify="right")
table.add_column("Updated", style="green")
for s in sessions:
table.add_row(
s.session_id[:8], # Truncate ID for readability
s.name or "(unnamed)",
s.actor_name or "(none)",
str(s.message_count),
s.updated_at.strftime("%Y-%m-%d %H:%M"),
)
console.print(table)
console.print()
# Reuse the summary dict already computed by _session_list_dict to avoid
# duplicating the total_msgs / sorted_sessions / most_recent / oldest logic.
summary = data["summary"]
summary_table = Table.grid(padding=(0, 1))
summary_table.add_column(style="cyan bold", justify="left")
summary_table.add_column(style="white", justify="left")
summary_table.add_row("Total:", str(summary["total"]))
summary_table.add_row("Most Recent:", summary["most_recent"] or "")
summary_table.add_row("Oldest:", summary["oldest"] or "")
summary_table.add_row("Total Messages:", str(summary["total_messages"]))
summary_table.add_row("Storage:", summary["storage"])
console.print(Panel(summary_table, title="Summary", border_style="blue"))
console.print()
console.print(f"[green]✓ OK[/green] {len(sessions)} sessions listed")
render_list_rich(sessions, data)
@app.command()
@@ -355,8 +202,6 @@ def show(
) -> None:
"""Show session details and recent messages.
Displays session metadata, recent messages, linked plans, and token usage.
Examples:
agents session show 01HXYZ...
agents session show 01HXYZ... --format json
@@ -370,77 +215,7 @@ def show(
typer.echo(format_output(dict(data), fmt))
return
# Session summary panel — field order per spec: ID, Actor, Messages,
# Created, Updated, Automation (docs/specification.md §agents session show)
details = (
f"[bold]ID:[/bold] {session.session_id}\n"
f"[bold]Actor:[/bold] {session.actor_name or '(none)'}\n"
f"[bold]Messages:[/bold] {session.message_count}\n"
f"[bold]Created:[/bold] {session.created_at.strftime('%Y-%m-%d %H:%M')}\n"
f"[bold]Updated:[/bold] {session.updated_at.strftime('%Y-%m-%d %H:%M')}\n"
f"[bold]Automation:[/bold] {session.automation or '(none)'}"
)
console.print(Panel(details, title="Session Summary", expand=False))
# Recent messages
if session.messages:
recent = session.messages[-5:]
msg_table = Table(title="Recent Messages", show_header=True)
msg_table.add_column("Role", style="cyan")
msg_table.add_column("Text")
for msg in recent:
text = msg.content
if len(text) > 80:
text = text[:77] + "..."
msg_table.add_row(msg.role.value, text)
console.print(msg_table)
# Linked plans — spec requires Plan ID / Phase / State columns
if session.linked_plans:
plan_table = Table(title="Linked Plans", show_header=True)
plan_table.add_column("Plan ID", style="cyan")
plan_table.add_column("Phase")
plan_table.add_column("State")
for lp in session.linked_plans:
plan_table.add_row(lp.plan_id, lp.phase, lp.state)
console.print(Panel(plan_table, title="Linked Plans", expand=False))
elif session.linked_plan_ids:
# Fallback: only flat IDs available
plan_text = "\n".join(f"{pid}" for pid in session.linked_plan_ids)
console.print(Panel(plan_text, title="Linked Plans", expand=False))
# Token usage
tu = session.token_usage
usage_text = (
f"[blue]Input Tokens:[/blue] {tu.input_tokens:,}\n"
f"[blue]Output Tokens:[/blue] {tu.output_tokens:,}\n"
f"[yellow]Estimated Cost:[/yellow] ${tu.estimated_cost:.4f}"
)
console.print(Panel(usage_text, title="Token Usage", expand=False))
# Cost budget (per-session budget cap, #584)
if session.cost_budget is not None:
cb = session.cost_budget
util = cb.utilization()
util_display = f"{util * 100:.1f}%" if util is not None else "N/A"
remaining = cb.remaining()
remaining_display = (
f"${remaining:.4f}" if remaining is not None else "unlimited"
)
max_display = (
f"${cb.max_cost_usd:.4f}"
if cb.max_cost_usd is not None
else "unlimited"
)
budget_text = (
f"[blue]Total Cost:[/blue] ${cb.total_cost:.4f}\n"
f"[blue]Max Cost:[/blue] {max_display}\n"
f"[yellow]Utilization:[/yellow] {util_display}\n"
f"[green]Remaining:[/green] {remaining_display}"
)
console.print(Panel(budget_text, title="Cost Budget", expand=False))
console.print("[green bold]✓ OK[/green bold] Session details loaded")
render_show_rich(session)
except SessionNotFoundError as exc:
console.print(f"[red]Session not found:[/red] {session_id}")
@@ -471,16 +246,12 @@ def delete(
) -> None:
"""Delete a session permanently.
A confirmation prompt is shown unless ``--yes`` is provided.
Examples:
agents session delete 01HXYZ...
agents session delete 01HXYZ... --yes
"""
try:
service = _get_session_service()
# Verify session exists before prompting and capture message count
session_to_delete = service.get(session_id)
message_count = session_to_delete.message_count
@@ -492,37 +263,9 @@ def delete(
service.delete(session_id)
# Rich output: render Deletion Summary and Cleanup panels
if fmt == OutputFormat.RICH:
# Deletion Summary panel
summary_table = Table.grid(padding=(0, 1))
summary_table.add_column(style="cyan bold", justify="left")
summary_table.add_column(style="white", justify="left")
summary_table.add_row("Session:", session_id)
summary_table.add_row("ID:", session_id)
summary_table.add_row("Messages:", f"{message_count} removed")
summary_table.add_row("Storage:", "0 KB freed")
summary_table.add_row("Plans Orphaned:", "0")
console.print(
Panel(summary_table, title="Deletion Summary", border_style="blue")
)
console.print()
# Cleanup panel
cleanup_table = Table.grid(padding=(0, 1))
cleanup_table.add_column(style="cyan bold", justify="left")
cleanup_table.add_column(style="white", justify="left")
cleanup_table.add_row("Backups:", "none")
cleanup_table.add_row("Logs:", "preserved")
cleanup_table.add_row("Context:", "cleared")
cleanup_table.add_row("Checkpoints:", "none")
console.print(Panel(cleanup_table, title="Cleanup", border_style="blue"))
console.print()
console.print("[green]✓ OK[/green] Session deleted")
render_delete_rich(session_id, message_count)
else:
# Non-rich formats: simple message
console.print(f"[green]✓ OK[/green] Session {session_id} deleted")
except SessionNotFoundError as exc:
@@ -561,15 +304,6 @@ def export_session(
) -> None:
"""Export a session as JSON or Markdown.
Writes the full session data to a file or stdout. The default format is
JSON (canonical, importable). Use ``--format md`` for a human-readable
Markdown transcript (lossy — cannot be re-imported).
On success, three Rich panels are displayed: "Session Export" (session ID,
output path, message count, file size, format), "Contents" (messages, plan
references, metadata keys, actor config, schema version), and "Integrity"
(checksum, encrypted flag).
Examples:
agents session export 01HXYZ...
agents session export 01HXYZ... -o session.json
@@ -582,30 +316,7 @@ def export_session(
try:
service = _get_session_service()
json_data: dict[str, Any] = {}
if fmt == "md":
# Markdown export: load full session with messages
session = service.get(session_id)
# Load messages via export_session to populate session.messages
json_data = service.export_session(session_id)
messages = [
SessionMessage(
message_id=m["message_id"],
role=MessageRole(m["role"]),
content=m["content"],
sequence=m["sequence"],
timestamp=m["timestamp"],
metadata=m.get("metadata", {}),
tool_call_id=m.get("tool_call_id"),
)
for m in json_data.get("messages", [])
]
session.messages = messages
content = session.as_export_markdown()
else:
json_data = service.export_session(session_id)
content = json.dumps(json_data, indent=2, default=str)
json_data, content, _ = build_export_content(service, session_id, fmt)
if output is not None:
if output.exists() and not force:
@@ -614,14 +325,12 @@ def export_session(
"Use --force to overwrite."
)
raise typer.Exit(1)
# Create parent directories if needed
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(content, encoding="utf-8")
else:
typer.echo(content)
# Render Rich panels for both file and stdout export paths
_render_export_panels(
render_export_panels(
session_id=session_id,
output=output,
content=content,
@@ -644,83 +353,6 @@ def export_session(
raise typer.Exit(1) from exc
def _render_export_panels(
*,
session_id: str,
output: Path | None,
content: str,
export_data: dict[str, Any],
fmt: str,
) -> None:
"""Render the three spec-required Rich panels for ``agents session export``.
Displays:
- "Session Export" panel: session ID, output path, message count, size, format
- "Contents" panel: messages, plan references, metadata keys, actor config,
schema version
- "Integrity" panel: checksum, encrypted flag
- ``✓ OK Export completed`` success line
"""
# Compute file size from content
size_bytes = len(content.encode("utf-8"))
if size_bytes < 1024:
size_str = f"{size_bytes} B"
elif size_bytes < 1024 * 1024:
size_str = f"{size_bytes // 1024} KB"
else:
size_str = f"{size_bytes // (1024 * 1024)} MB"
# Derive display values from export data (JSON format) or defaults (md)
message_count = len(export_data.get("messages", []))
plan_refs = len(export_data.get("linked_plan_ids", []))
metadata_keys = len(export_data.get("metadata", {}))
actor_config = "included" if export_data.get("actor_name") else "none"
schema_version = export_data.get("schema_version", "v1")
checksum_raw = export_data.get("checksum", "")
checksum_display = (
f"sha256:{checksum_raw[:4]}...{checksum_raw[-4:]}"
if len(checksum_raw) >= 8
else checksum_raw or "n/a"
)
output_display = str(output) if output is not None else "(stdout)"
format_display = "JSON" if fmt == "json" else "Markdown"
# Session Export panel
export_table = Table.grid(padding=(0, 1))
export_table.add_column(style="cyan bold", justify="left")
export_table.add_column(style="white", justify="left")
export_table.add_row("Session:", session_id)
export_table.add_row("Output:", output_display)
export_table.add_row("Messages:", str(message_count))
export_table.add_row("Size:", size_str)
export_table.add_row("Format:", format_display)
console.print(Panel(export_table, title="Session Export", border_style="blue"))
console.print()
# Contents panel
contents_table = Table.grid(padding=(0, 1))
contents_table.add_column(style="blue bold", justify="left")
contents_table.add_column(style="white", justify="left")
contents_table.add_row("Messages:", str(message_count))
contents_table.add_row("Plan References:", str(plan_refs))
contents_table.add_row("Metadata Keys:", str(metadata_keys))
contents_table.add_row("Actor Config:", actor_config)
contents_table.add_row("Schema Version:", str(schema_version))
console.print(Panel(contents_table, title="Contents", border_style="blue"))
console.print()
# Integrity panel
integrity_table = Table.grid(padding=(0, 1))
integrity_table.add_column(style="magenta bold", justify="left")
integrity_table.add_column(style="white", justify="left")
integrity_table.add_row("Checksum:", checksum_display)
integrity_table.add_row("Encrypted:", "no")
console.print(Panel(integrity_table, title="Integrity", border_style="blue"))
console.print()
console.print("[green]✓ OK[/green] Export completed")
@app.command("import")
def import_session(
input_file: Annotated[
@@ -730,9 +362,6 @@ def import_session(
) -> None:
"""Import a session from a JSON file.
The file must have been produced by ``agents session export`` and must
contain a valid schema version and checksum.
Examples:
agents session import -i session.json
"""
@@ -753,29 +382,7 @@ def import_session(
actor_name = data.get("actor_name")
session = service.import_session(data)
# Session Import panel
session_details = (
f"[bold]Input:[/bold] {input_file}\n"
f"[bold]Session ID:[/bold] {session.session_id}\n"
f"[bold]Messages:[/bold] {session.message_count}\n"
f"[bold]Schema:[/bold] {schema_version}"
)
console.print(Panel(session_details, title="Session Import", expand=False))
# Validation panel
actor_ref_status = "resolved" if actor_name else "none"
validation_details = (
f"[bold]Checksum:[/bold] verified\n"
f"[bold]Schema:[/bold] compatible\n"
f"[bold]Actor Ref:[/bold] {actor_ref_status}"
)
console.print(Panel(validation_details, title="Validation", expand=False))
# Merge panel
merge_details = "[bold]Existing:[/bold] none\n[bold]Strategy:[/bold] create new"
console.print(Panel(merge_details, title="Merge", expand=False))
console.print("[green]✓ OK[/green] Import completed")
render_import_rich(session, input_file, schema_version, actor_name)
except SessionImportError as exc:
console.print(f"[red]Import error:[/red] {exc}")
@@ -807,11 +414,15 @@ def tell(
bool,
typer.Option("--stream", help="Stream response in real-time"),
] = False,
fmt: Annotated[
str,
typer.Option("--format", 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.
Renders spec-required Rich panels (Plan Request, Commands Executed, Result,
Usage). Supports ``--format json/yaml/plain/table/color``.
Examples:
agents session tell --session 01HXYZ... "Hello, world"
@@ -820,15 +431,14 @@ def tell(
"""
try:
service = _get_session_service()
session_obj = service.get(session_id)
# Append user message
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
@@ -840,17 +450,40 @@ def tell(
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
actor_display = actor or session_obj.actor_name or "local/orchestrator"
automation = session_obj.automation or "default"
payload = build_tell_payload(
prompt=prompt,
session_id=session_id,
actor_display=actor_display,
automation=automation,
stream=stream,
assistant_content=assistant_content if stream else None,
)
console.print(f"[dim]user:[/dim] {escape(prompt)}")
console.print(f"[cyan]assistant:[/cyan] {escape(assistant_content)}")
status_text = "✓ OK Orchestrator acknowledged request"
command_str = build_tell_command(
prompt=prompt,
session_id=session_id,
actor=actor or session_obj.actor_name,
stream=stream,
)
envelope_messages = [
{
"level": "ok",
"text": "Orchestrator acknowledged request; no commands were executed.",
}
]
render_tell_output(
payload=payload,
status_text=status_text,
command_str=command_str,
envelope_messages=envelope_messages,
fmt=fmt,
stream=stream,
assistant_content=assistant_content,
)
except SessionNotFoundError as exc:
console.print(f"[red]Session not found:[/red] {session_id}")
@@ -0,0 +1,352 @@
"""Shared helper utilities for the Session CLI commands.
Extracted from ``session.py`` to keep that file within the 500-line project
limit. All public symbols here are imported by ``session.py`` and are not
part of the public package API.
"""
from __future__ import annotations
import json
from collections import OrderedDict
from pathlib import Path
from typing import Any
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from cleveragents.domain.models.core.session import MessageRole, Session, SessionMessage
console = Console()
def session_summary_dict(session: Session) -> OrderedDict[str, Any]:
"""Build a stable-ordered summary dict for a session."""
result: OrderedDict[str, Any] = OrderedDict()
result["session_id"] = session.session_id
result["actor"] = session.actor_name or "(none)"
result["namespace"] = session.namespace
result["messages"] = session.message_count
result["created"] = session.created_at.isoformat()
result["updated"] = session.updated_at.isoformat()
return result
def session_list_dict(sessions: list[Session]) -> dict[str, Any]:
"""Build the list output with summary stats per spec."""
items = []
for s in sessions:
items.append(
{
"id": s.session_id,
"name": s.name or None,
"actor": s.actor_name or "(none)",
"messages": s.message_count,
"updated": s.updated_at.isoformat(),
}
)
# Build summary section per spec
total_messages = sum(s.message_count for s in sessions)
# Find most recent and oldest sessions
if sessions:
sorted_sessions = sorted(sessions, key=lambda x: x.updated_at, reverse=True)
most_recent = sorted_sessions[0].name or sorted_sessions[0].session_id[:8]
oldest = sorted_sessions[-1].name or sorted_sessions[-1].session_id[:8]
else:
most_recent = None
oldest = None
summary = {
"total": len(sessions),
"most_recent": most_recent,
"oldest": oldest,
"total_messages": total_messages,
"storage": "0 KB", # Placeholder - actual storage calculation not implemented
}
return {
"sessions": items,
"summary": summary,
}
def render_export_panels(
*,
session_id: str,
output: Path | None,
content: str,
export_data: dict[str, Any],
fmt: str,
) -> None:
"""Render the three spec-required Rich panels for ``agents session export``.
Displays:
- "Session Export" panel: session ID, output path, message count, size, format
- "Contents" panel: messages, plan references, metadata keys, actor config,
schema version
- "Integrity" panel: checksum, encrypted flag
- `` OK Export completed`` success line
"""
# Compute file size from content
size_bytes = len(content.encode("utf-8"))
if size_bytes < 1024:
size_str = f"{size_bytes} B"
elif size_bytes < 1024 * 1024:
size_str = f"{size_bytes // 1024} KB"
else:
size_str = f"{size_bytes // (1024 * 1024)} MB"
# Derive display values from export data (JSON format) or defaults (md)
message_count = len(export_data.get("messages", []))
plan_refs = len(export_data.get("linked_plan_ids", []))
metadata_keys = len(export_data.get("metadata", {}))
actor_config = "included" if export_data.get("actor_name") else "none"
schema_version = export_data.get("schema_version", "v1")
checksum_raw = export_data.get("checksum", "")
checksum_display = (
f"sha256:{checksum_raw[:4]}...{checksum_raw[-4:]}"
if len(checksum_raw) >= 8
else checksum_raw or "n/a"
)
output_display = str(output) if output is not None else "(stdout)"
format_display = "JSON" if fmt == "json" else "Markdown"
# Session Export panel
export_table = Table.grid(padding=(0, 1))
export_table.add_column(style="cyan bold", justify="left")
export_table.add_column(style="white", justify="left")
export_table.add_row("Session:", session_id)
export_table.add_row("Output:", output_display)
export_table.add_row("Messages:", str(message_count))
export_table.add_row("Size:", size_str)
export_table.add_row("Format:", format_display)
console.print(Panel(export_table, title="Session Export", border_style="blue"))
console.print()
# Contents panel
contents_table = Table.grid(padding=(0, 1))
contents_table.add_column(style="blue bold", justify="left")
contents_table.add_column(style="white", justify="left")
contents_table.add_row("Messages:", str(message_count))
contents_table.add_row("Plan References:", str(plan_refs))
contents_table.add_row("Metadata Keys:", str(metadata_keys))
contents_table.add_row("Actor Config:", actor_config)
contents_table.add_row("Schema Version:", str(schema_version))
console.print(Panel(contents_table, title="Contents", border_style="blue"))
console.print()
# Integrity panel
integrity_table = Table.grid(padding=(0, 1))
integrity_table.add_column(style="magenta bold", justify="left")
integrity_table.add_column(style="white", justify="left")
integrity_table.add_row("Checksum:", checksum_display)
integrity_table.add_row("Encrypted:", "no")
console.print(Panel(integrity_table, title="Integrity", border_style="blue"))
console.print()
console.print("[green]✓ OK[/green] Export completed")
def render_show_rich(session: Session) -> None:
"""Render the Rich panels for ``agents session show``."""
details = (
f"[bold]ID:[/bold] {session.session_id}\n"
f"[bold]Actor:[/bold] {session.actor_name or '(none)'}\n"
f"[bold]Messages:[/bold] {session.message_count}\n"
f"[bold]Created:[/bold] {session.created_at.strftime('%Y-%m-%d %H:%M')}\n"
f"[bold]Updated:[/bold] {session.updated_at.strftime('%Y-%m-%d %H:%M')}\n"
f"[bold]Automation:[/bold] {session.automation or '(none)'}"
)
console.print(Panel(details, title="Session Summary", expand=False))
if session.messages:
recent = session.messages[-5:]
msg_table = Table(title="Recent Messages", show_header=True)
msg_table.add_column("Role", style="cyan")
msg_table.add_column("Text")
for msg in recent:
text = msg.content
if len(text) > 80:
text = text[:77] + "..."
msg_table.add_row(msg.role.value, text)
console.print(msg_table)
if session.linked_plans:
plan_table = Table(title="Linked Plans", show_header=True)
plan_table.add_column("Plan ID", style="cyan")
plan_table.add_column("Phase")
plan_table.add_column("State")
for lp in session.linked_plans:
plan_table.add_row(lp.plan_id, lp.phase, lp.state)
console.print(Panel(plan_table, title="Linked Plans", expand=False))
elif session.linked_plan_ids:
plan_text = "\n".join(f"{pid}" for pid in session.linked_plan_ids)
console.print(Panel(plan_text, title="Linked Plans", expand=False))
tu = session.token_usage
usage_text = (
f"[blue]Input Tokens:[/blue] {tu.input_tokens:,}\n"
f"[blue]Output Tokens:[/blue] {tu.output_tokens:,}\n"
f"[yellow]Estimated Cost:[/yellow] ${tu.estimated_cost:.4f}"
)
console.print(Panel(usage_text, title="Token Usage", expand=False))
if session.cost_budget is not None:
cb = session.cost_budget
util = cb.utilization()
util_display = f"{util * 100:.1f}%" if util is not None else "N/A"
remaining = cb.remaining()
remaining_display = (
f"${remaining:.4f}" if remaining is not None else "unlimited"
)
max_display = (
f"${cb.max_cost_usd:.4f}" if cb.max_cost_usd is not None else "unlimited"
)
budget_text = (
f"[blue]Total Cost:[/blue] ${cb.total_cost:.4f}\n"
f"[blue]Max Cost:[/blue] {max_display}\n"
f"[yellow]Utilization:[/yellow] {util_display}\n"
f"[green]Remaining:[/green] {remaining_display}"
)
console.print(Panel(budget_text, title="Cost Budget", expand=False))
console.print("[green bold]✓ OK[/green bold] Session details loaded")
def render_list_rich(sessions: list[Session], data: dict[str, Any]) -> None:
"""Render the Rich table and summary for ``agents session list``."""
table = Table(title="Sessions", show_header=True, border_style="blue")
table.add_column("ID", style="cyan")
table.add_column("Name", style="magenta")
table.add_column("Actor", style="blue")
table.add_column("Messages", justify="right")
table.add_column("Updated", style="green")
for s in sessions:
table.add_row(
s.session_id[:8],
s.name or "(unnamed)",
s.actor_name or "(none)",
str(s.message_count),
s.updated_at.strftime("%Y-%m-%d %H:%M"),
)
console.print(table)
console.print()
summary = data["summary"]
summary_table = Table.grid(padding=(0, 1))
summary_table.add_column(style="cyan bold", justify="left")
summary_table.add_column(style="white", justify="left")
summary_table.add_row("Total:", str(summary["total"]))
summary_table.add_row("Most Recent:", summary["most_recent"] or "")
summary_table.add_row("Oldest:", summary["oldest"] or "")
summary_table.add_row("Total Messages:", str(summary["total_messages"]))
summary_table.add_row("Storage:", summary["storage"])
console.print(Panel(summary_table, title="Summary", border_style="blue"))
console.print()
console.print(f"[green]✓ OK[/green] {len(sessions)} sessions listed")
def render_create_rich(session: Session) -> None:
"""Render the Rich panels for ``agents session create``."""
details = (
f"[bold]Session ID:[/bold] {session.session_id}\n"
f"[bold]Actor:[/bold] {session.actor_name or '(none)'}\n"
f"[bold]Namespace:[/bold] {session.namespace}\n"
f"[bold]Created:[/bold] {session.created_at.strftime('%Y-%m-%d %H:%M')}"
)
console.print(Panel(details, title="Session", expand=False))
settings_text = (
"[yellow]Automation:[/yellow] default\n"
"[yellow]Streaming:[/yellow] off\n"
"[yellow]Context:[/yellow] default\n"
"[yellow]Memory:[/yellow] enabled\n"
"[yellow]Max History:[/yellow] 50 turns"
)
console.print(Panel(settings_text, title="Settings", expand=False))
console.print("[green]✓ OK[/green] Session created")
def render_delete_rich(session_id: str, message_count: int) -> None:
"""Render the Rich panels for ``agents session delete``."""
summary_table = Table.grid(padding=(0, 1))
summary_table.add_column(style="cyan bold", justify="left")
summary_table.add_column(style="white", justify="left")
summary_table.add_row("Session:", session_id)
summary_table.add_row("ID:", session_id)
summary_table.add_row("Messages:", f"{message_count} removed")
summary_table.add_row("Storage:", "0 KB freed")
summary_table.add_row("Plans Orphaned:", "0")
console.print(Panel(summary_table, title="Deletion Summary", border_style="blue"))
console.print()
cleanup_table = Table.grid(padding=(0, 1))
cleanup_table.add_column(style="cyan bold", justify="left")
cleanup_table.add_column(style="white", justify="left")
cleanup_table.add_row("Backups:", "none")
cleanup_table.add_row("Logs:", "preserved")
cleanup_table.add_row("Context:", "cleared")
cleanup_table.add_row("Checkpoints:", "none")
console.print(Panel(cleanup_table, title="Cleanup", border_style="blue"))
console.print()
console.print("[green]✓ OK[/green] Session deleted")
def render_import_rich(
session: Session, input_file: Any, schema_version: str, actor_name: str | None
) -> None:
"""Render the Rich panels for ``agents session import``."""
session_details = (
f"[bold]Input:[/bold] {input_file}\n"
f"[bold]Session ID:[/bold] {session.session_id}\n"
f"[bold]Messages:[/bold] {session.message_count}\n"
f"[bold]Schema:[/bold] {schema_version}"
)
console.print(Panel(session_details, title="Session Import", expand=False))
actor_ref_status = "resolved" if actor_name else "none"
validation_details = (
f"[bold]Checksum:[/bold] verified\n"
f"[bold]Schema:[/bold] compatible\n"
f"[bold]Actor Ref:[/bold] {actor_ref_status}"
)
console.print(Panel(validation_details, title="Validation", expand=False))
merge_details = "[bold]Existing:[/bold] none\n[bold]Strategy:[/bold] create new"
console.print(Panel(merge_details, title="Merge", expand=False))
console.print("[green]✓ OK[/green] Import completed")
def build_export_content(
service: Any, session_id: str, fmt: str
) -> tuple[dict[str, Any], str, Session | None]:
"""Build the export content string and data dict.
Returns ``(json_data, content, session_or_none)``.
"""
Review

BLOCKING — In-function import violates project import rules

import json must not appear inside function bodies. CONTRIBUTING.md requires all imports at the top of the file.

Fix: Move import json to the top of session_helpers.py:

# At the top of session_helpers.py:
import json

Then remove this in-function import.


Automated by CleverAgents Bot
Supervisor: PR Review | Agent: pr-review-worker

**BLOCKING — In-function import violates project import rules** `import json` must not appear inside function bodies. CONTRIBUTING.md requires all imports at the top of the file. **Fix:** Move `import json` to the top of `session_helpers.py`: ```python # At the top of session_helpers.py: import json ``` Then remove this in-function import. --- Automated by CleverAgents Bot Supervisor: PR Review | Agent: pr-review-worker
if fmt == "md":
session = service.get(session_id)
json_data: dict[str, Any] = service.export_session(session_id)
messages = [
SessionMessage(
message_id=m["message_id"],
role=MessageRole(m["role"]),
content=m["content"],
sequence=m["sequence"],
timestamp=m["timestamp"],
metadata=m.get("metadata", {}),
tool_call_id=m.get("tool_call_id"),
)
for m in json_data.get("messages", [])
]
session.messages = messages
return json_data, session.as_export_markdown(), session
else:
json_data = service.export_session(session_id)
return json_data, json.dumps(json_data, indent=2, default=str), None
@@ -0,0 +1,292 @@
"""Tell-command helpers for the Session CLI.
Extracted from ``session.py`` to keep that file within the 500-line project
limit. All public symbols here are imported by ``session.py`` and are not
part of the public package API.
"""
from __future__ import annotations
import shlex
import sys
import textwrap
from typing import Any
import typer
from rich.console import Console
from rich.panel import Panel
from cleveragents.cli.formatting import OutputFormat, format_output
console = Console()
# Reusable --format option description (mirrors session.py constant)
_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
def truncate_prompt(prompt: str, max_length: int = 72) -> str:
"""Return a prompt trimmed to ``max_length`` characters with ellipsis."""
text = prompt.strip()
if len(text) <= max_length:
return text
# ``textwrap.shorten`` collapses internal whitespace and adds the placeholder.
return textwrap.shorten(text, width=max_length, placeholder="...")
def build_tell_command(
*, prompt: str, session_id: str, actor: str | None, stream: bool
) -> str:
"""Construct the canonical command string for envelope metadata."""
parts: list[str] = ["agents", "session", "tell"]
if stream:
parts.append("--stream")
parts.extend(["--session", session_id])
if actor:
parts.extend(["--actor", actor])
parts.append(shlex.quote(prompt))
return " ".join(parts)
def build_tell_payload(
*,
prompt: str,
session_id: str,
actor_display: str,
automation: str,
stream: bool,
assistant_content: str | None = None,
) -> dict[str, Any]:
"""Assemble the structured payload for ``session tell`` output."""
plan_request: dict[str, Any] = {
"actor": actor_display,
"session": session_id,
"prompt": truncate_prompt(prompt, max_length=96),
}
if stream:
plan_request["mode"] = "streaming"
else:
plan_request["automation"] = automation
result_block: dict[str, Any] = {
"action": None,
"project": None,
"resource": None,
"note": "Orchestrator has not executed commands yet.",
}
usage_block: dict[str, Any] = {
"input_tokens": 0,
"output_tokens": 0,
"cost": 0.0,
"duration_s": 0.0,
"tool_calls": 0,
}
payload: dict[str, Any] = {
"plan_request": plan_request,
"commands_executed": [],
"result": result_block,
"usage": usage_block,
}
if assistant_content is not None:
payload["assistant_response"] = assistant_content
return payload
def format_tell_plain(
payload: dict[str, Any],
status_text: str,
*,
stream: bool = False,
assistant_content: str | None = None,
) -> str:
"""Render the tell payload in plain-text format."""
lines: list[str] = []
plan = payload["plan_request"]
title = "Session" if stream else "Plan Request"
lines.append(title)
lines.append(f" Actor: {plan['actor']}")
lines.append(f" Session: {plan['session']}")
if stream:
lines.append(" Mode: streaming")
else:
lines.append(f" Automation: {plan['automation']}")
lines.append(f" Prompt: {plan['prompt']}")
lines.append("")
if not stream:
lines.append("Commands Executed")
commands = payload["commands_executed"]
if commands:
for command in commands:
lines.append(f" - {command}")
else:
lines.append(" (none)")
lines.append("")
lines.append("Result")
result = payload.get("result") or {}
included_any = False
for key, label in (
("action", "Action"),
("project", "Project"),
("resource", "Resource"),
):
value = result.get(key)
if value:
lines.append(f" {label}: {value}")
included_any = True
note = result.get("note")
if note:
lines.append(f" Note: {note}")
included_any = True
if not included_any:
lines.append(" (pending)")
lines.append("")
elif assistant_content:
lines.append("Assistant Response")
lines.append(f" {assistant_content}")
lines.append("")
usage = payload["usage"]
lines.append("Usage")
lines.append(f" Input Tokens: {usage['input_tokens']}")
lines.append(f" Output Tokens: {usage['output_tokens']}")
lines.append(f" Cost: ${usage['cost']:.4f}")
lines.append(f" Duration: {usage['duration_s']:.1f}s")
lines.append(f" Tool Calls: {usage['tool_calls']}")
lines.append("")
lines.append(status_text)
return "\n".join(lines).rstrip()
def render_tell_rich(payload: dict[str, Any], status_text: str) -> None:
"""Render the tell payload using Rich panels."""
plan = payload["plan_request"]
plan_lines = [
f"[magenta]Actor:[/magenta] {plan['actor']}",
f"[cyan]Session:[/cyan] {plan['session']}",
]
if "automation" in plan:
plan_lines.append(f"[yellow]Automation:[/yellow] {plan['automation']}")
plan_lines.append(f"[blue]Prompt:[/blue] {plan['prompt']}")
console.print(Panel("\n".join(plan_lines), title="Plan Request", expand=False))
commands = payload["commands_executed"]
if commands:
commands_text = "\n".join(f"- {command}" for command in commands)
else:
commands_text = "(none)"
console.print(Panel(commands_text, title="Commands Executed", expand=False))
result = payload.get("result") or {}
result_lines: list[str] = []
for key, label in (
("action", "Action"),
("project", "Project"),
("resource", "Resource"),
):
value = result.get(key)
if value:
result_lines.append(f"[green]{label}:[/green] {value}")
note = result.get("note")
if note:
result_lines.append(note)
if not result_lines:
result_lines.append("(pending orchestrator integration)")
console.print(Panel("\n".join(result_lines), title="Result", expand=False))
usage = payload["usage"]
usage_lines = [
f"[blue]Input Tokens:[/blue] {usage['input_tokens']}",
f"[blue]Output Tokens:[/blue] {usage['output_tokens']}",
f"[yellow]Cost:[/yellow] ${usage['cost']:.4f}",
f"[yellow]Duration:[/yellow] {usage['duration_s']:.1f}s",
f"[yellow]Tool Calls:[/yellow] {usage['tool_calls']}",
]
console.print(Panel("\n".join(usage_lines), title="Usage", expand=False))
console.print(f"[green]{status_text}[/green]")
def render_tell_streaming(
payload: dict[str, Any], assistant_content: str, status_text: str
) -> None:
"""Render streaming output matching the spec stub behaviour."""
plan = payload["plan_request"]
plan_lines = [
f"[magenta]Actor:[/magenta] {plan['actor']}",
f"[cyan]Session:[/cyan] {plan['session']}",
"[yellow]Mode:[/yellow] streaming",
f"[blue]Prompt:[/blue] {plan['prompt']}",
]
console.print(Panel("\n".join(plan_lines), title="Session", expand=False))
for char in assistant_content:
sys.stdout.write(char)
sys.stdout.flush()
sys.stdout.write("\n")
sys.stdout.flush()
usage = payload["usage"]
usage_lines = [
f"[blue]Input Tokens:[/blue] {usage['input_tokens']}",
f"[blue]Output Tokens:[/blue] {usage['output_tokens']}",
f"[yellow]Cost:[/yellow] ${usage['cost']:.4f}",
f"[yellow]Duration:[/yellow] {usage['duration_s']:.1f}s",
f"[yellow]Tool Calls:[/yellow] {usage['tool_calls']}",
]
console.print(Panel("\n".join(usage_lines), title="Usage", expand=False))
console.print(f"[green]{status_text}[/green]")
def render_tell_output(
*,
payload: dict[str, Any],
status_text: str,
command_str: str,
envelope_messages: list[dict[str, Any]],
fmt: str,
stream: bool,
assistant_content: str,
) -> None:
"""Dispatch tell output to the appropriate renderer based on ``fmt``."""
fmt_lower = fmt.lower()
if fmt_lower in (OutputFormat.JSON.value, OutputFormat.YAML.value):
Review

BLOCKING — In-function import violates project import rules

import typer must not appear inside function bodies. CONTRIBUTING.md requires all imports at the top of the file (the sole exception is if TYPE_CHECKING: blocks).

Fix: Move import typer to the top of session_tell.py and remove this in-function import.

# At the top of session_tell.py:
import typer

This is the same issue as was fixed in session.py in the prior commit — the refactoring accidentally introduced it again in the extracted module.


Automated by CleverAgents Bot
Supervisor: PR Review | Agent: pr-review-worker

**BLOCKING — In-function import violates project import rules** `import typer` must not appear inside function bodies. CONTRIBUTING.md requires all imports at the top of the file (the sole exception is `if TYPE_CHECKING:` blocks). **Fix:** Move `import typer` to the top of `session_tell.py` and remove this in-function import. ```python # At the top of session_tell.py: import typer ``` This is the same issue as was fixed in `session.py` in the prior commit — the refactoring accidentally introduced it again in the extracted module. --- Automated by CleverAgents Bot Supervisor: PR Review | Agent: pr-review-worker
typer.echo(
format_output(
payload,
fmt_lower,
command=command_str,
status="ok",
exit_code=0,
messages=envelope_messages,
)
)
return
if fmt_lower in (OutputFormat.PLAIN.value, OutputFormat.TABLE.value):
typer.echo(
format_tell_plain(
payload,
status_text,
stream=stream,
assistant_content=assistant_content if stream else None,
)
)
return
if fmt_lower == OutputFormat.COLOR.value:
if stream:
render_tell_streaming(payload, assistant_content, status_text)
else:
render_tell_rich(payload, status_text)
return
if stream:
render_tell_streaming(payload, assistant_content, status_text)
else:
render_tell_rich(payload, status_text)