generated from coulomb/repo-seed
146 lines
5.1 KiB
Python
146 lines
5.1 KiB
Python
"""Assistance orchestrator (T06).
|
|
|
|
The piece that turns raw user intent + collected context into a well-formed
|
|
request for the LLM adapter (T04), then turns the adapter response into the
|
|
final terminal output the user sees.
|
|
|
|
Responsibilities in this slice:
|
|
- Own the end-to-end happy path after Typer argument parsing.
|
|
- Coordinate context collector (T02), risk classifier (T03), and LLMAdapter (T04).
|
|
- Keep the CLI surface (main.py) thin — it should only do argument parsing,
|
|
help/version, and delegation to this orchestrator.
|
|
- Be testable in isolation with the FakeLLMAdapter (critical for T07).
|
|
|
|
This module is the natural home for future prompt framing, context packing
|
|
with token awareness, safety charter injection, and response post-processing.
|
|
|
|
See workplan CYA-WP-0001-T06.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from rich.console import Console
|
|
from rich.panel import Panel
|
|
|
|
from cya.context.collector import collect, render_explanation
|
|
from cya.memory import recall_preferences
|
|
from cya.safety.risk import classify, get_user_confirmation
|
|
from cya.llm.adapter import AssistanceRequest, FakeLLMAdapter
|
|
|
|
|
|
console = Console()
|
|
|
|
|
|
def handle_request(
|
|
user_request: str,
|
|
*,
|
|
explain_context: bool = False,
|
|
dry_run: bool = False,
|
|
) -> None:
|
|
"""Primary orchestrator entry point.
|
|
|
|
This is what the CLI (and future tests / other front-ends) should call.
|
|
It coordinates the full current flow:
|
|
context → safety (with mandatory confirmation) → LLMAdapter → render
|
|
"""
|
|
# 1. Context (always cheap; needed for safety "affected" and for the adapter)
|
|
try:
|
|
envelope = collect(".")
|
|
except Exception:
|
|
envelope = None
|
|
|
|
if explain_context and envelope:
|
|
try:
|
|
explanation = render_explanation(envelope)
|
|
console.print(
|
|
Panel(
|
|
explanation,
|
|
title="Context Envelope (T02)",
|
|
border_style="green",
|
|
padding=(1, 1),
|
|
)
|
|
)
|
|
except Exception as exc:
|
|
console.print(f"[red]Context explanation error: {exc}[/red]")
|
|
|
|
# T03 (memory wiring): consult after context (so safety can see it in future T04 0002),
|
|
# before risk/LLM. Real T02 prefs now available; graceful.
|
|
memory = {}
|
|
try:
|
|
memory = recall_preferences(".")
|
|
except Exception:
|
|
memory = {"error": "recall failed (graceful degradation)"}
|
|
|
|
if explain_context and memory.get("items"):
|
|
try:
|
|
prov = memory.get("provenance", [{}])[0]
|
|
console.print(
|
|
Panel(
|
|
f"Phase: {memory.get('phase')} | {len(memory.get('items', []))} items | {prov.get('source', 'local')}",
|
|
title="Memory Consulted (T03)",
|
|
border_style="blue",
|
|
padding=(0, 1),
|
|
)
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
# 2. Risk classification + mandatory confirmation (T03 safety; T04 memory signals)
|
|
assessment = classify(user_request, envelope, memory=memory)
|
|
|
|
if assessment.requires_confirmation:
|
|
from rich.table import Table
|
|
|
|
table = Table(
|
|
title=f"Risk Assessment — {assessment.level.value.upper()}",
|
|
show_header=False,
|
|
border_style="red",
|
|
)
|
|
table.add_row("Rationale", assessment.rationale)
|
|
if assessment.preview:
|
|
table.add_row("Preview", assessment.preview)
|
|
if assessment.affected_summary:
|
|
table.add_row("Would affect", assessment.affected_summary)
|
|
table.add_row("Rules", ", ".join(assessment.rules_triggered[:3]))
|
|
console.print(table)
|
|
|
|
if not get_user_confirmation(assessment):
|
|
console.print("[yellow]Action cancelled by user. No changes made.[/yellow]")
|
|
return
|
|
|
|
if dry_run:
|
|
console.print("[green]--dry-run acknowledged.[/green] No side-effects.")
|
|
return
|
|
|
|
# 3. Call through the single LLMAdapter boundary (T04)
|
|
adapter = FakeLLMAdapter()
|
|
ctx = (envelope.to_dict() if envelope else {}) or {}
|
|
ctx["memory"] = memory # T03: memory now in context passed to LLM (for personalization + explain)
|
|
llm_request = AssistanceRequest(
|
|
user_request=user_request,
|
|
context=ctx,
|
|
)
|
|
llm_response = adapter.complete(llm_request)
|
|
|
|
# 4. Render final user-facing artifact (T06 responsibility; T03 memory surface)
|
|
mem_line = ""
|
|
if memory.get("items"):
|
|
mem_line = f"\n[dim]Memory: {len(memory.get('items', []))} prefs (phase {memory.get('phase')}, {memory.get('provenance', [{}])[0].get('source', 'local')})[/dim]"
|
|
console.print(
|
|
Panel(
|
|
f"[bold]Suggestion:[/bold]\n{llm_response.suggestion}\n\n"
|
|
f"[dim]{llm_response.explanation}\n"
|
|
f"Rationale: {llm_response.rationale}{mem_line}[/dim]",
|
|
title="LLM Response (via T04 seam)",
|
|
border_style="magenta",
|
|
padding=(1, 1),
|
|
)
|
|
)
|
|
|
|
console.print(
|
|
"[green]✓[/green] Request processed by orchestrator (T02+T03+T04 coordinated by T06)."
|
|
)
|
|
|
|
|
|
__all__ = ["handle_request"]
|