Agents and tools

#[agent]
impl Agent for DealsAssistant {
    const NAME: &'static str = "deals-assistant";

    fn instructions(&self, run: &AgentRun) -> String {
        "You are the sales assistant of a small CRM team. Use the tools; never guess.".into()
    }

    /// Finds the deals the user can see, by words in the title or by stage.
    #[tool(read_only)]
    async fn search_deals(&self, run: &AgentRun, args: SearchDeals) -> Result<Vec<DealResource>> {
        // calls DealService with the user's access
    }
}

use crm_activities::{ActivityFilter, ActivityResource, ActivityService, CreateActivity};
use crm_core::{Activity, ActivityKind, Deal, Move, Stage, SubjectType};
use navree::ai::{Agent, AgentOptions, AgentRun, agent};
use navree::prelude::*;

use crate::{DealFilter, DealResource, DealService, MoveDeal};

/// The deals assistant in the CRM's chat panel.
#[injectable]
pub struct DealsAssistant {
    deals: Arc<DealService>,
    activities: Arc<ActivityService>,
}

#[agent]
impl Agent for DealsAssistant {
    const NAME: &'static str = "deals-assistant";

    fn instructions(&self, run: &AgentRun) -> String {
        format!(
            "You are the sales assistant of a small CRM team, talking to {name}. \
             Today is {today}. Amounts are in cents; show them in the deal's currency. \
             Look things up with the tools before you answer, and never invent deals, \
             people, dates or amounts. When a tool refuses, say so plainly.",
            name = run.user().name,
            today = run.now().date_naive(),
        )
    }

    fn options(&self) -> AgentOptions {
        AgentOptions {
            model: Some("smart".into()),
            max_steps: 8,
            ..AgentOptions::default()
        }
    }

    // tools: see below
}

impl DealsModule {
    fn wire(&self, m: &mut ModuleBuilder) {
        m.listen::<DealAssigned, NotifyDealOwner>()
            .on::<DealWon, NotifyDealOwner>()
            .templates(notification::TEMPLATES)
            .agent::<DealsAssistant>();
    }
}

#[agent]
impl Agent for DealsAssistant {
    // ...

    /// Finds the deals the user can see, newest first: by words in the title,
    /// by stage, or only the user's own. Returns at most 20.
    #[tool(read_only)]
    async fn search_deals(&self, run: &AgentRun, args: SearchDeals) -> Result<Vec<DealResource>> {
        let filter = DealFilter {
            search: args.query,
            stage: args.stage,
            owner_id: args.mine.then(|| run.user_id()),
            ..DealFilter::default()
        };
        let page = PageQuery { per_page: 20, ..PageQuery::default() };
        let deals = self
            .deals
            .list(run.uow(), run.accessible::<Deal, Read>(), filter, page)
            .await?;
        Ok(deals.data)
    }

    /// The activity on one deal, newest first: notes, calls, emails, meetings and tasks.
    #[tool(read_only)]
    async fn deal_activity(&self, run: &AgentRun, args: DealRef) -> Result<Vec<ActivityResource>> {
        let filter = ActivityFilter {
            subject_type: Some(SubjectType::Deal),
            subject_id: Some(args.deal_id),
            ..ActivityFilter::default()
        };
        let page = PageQuery { per_page: 30, ..PageQuery::default() };
        let activities = self
            .activities
            .list(run.uow(), run.accessible::<Activity, Read>(), filter, page)
            .await?;
        Ok(activities.data)
    }

    /// Logs a note, call, email or meeting on a deal, or creates a task with a due date.
    #[tool]
    #[can(Create, Activity)]
    async fn log_activity(&self, run: &AgentRun, args: LogActivity) -> Result<ActivityResource> {
        self.activities
            .create(
                run.uow(),
                run.access(),
                CreateActivity {
                    kind: args.kind,
                    subject: args.subject,
                    body: args.body,
                    due_at: args.due_at,
                    subject_type: SubjectType::Deal,
                    subject_id: args.deal_id,
                    assignee_id: None,
                },
            )
            .await
    }
}

use chrono::{DateTime, Utc};
use garde::Validate;
use serde::Deserialize;
use utoipa::ToSchema;
use uuid::Uuid;

#[derive(Debug, Deserialize, ToSchema, Validate)]
pub struct SearchDeals {
    /// Words from the deal title, e.g. "Acme renewal". Leave out to list every deal.
    #[garde(length(chars, max = 100))]
    pub query: Option<String>,
    /// Only deals in this stage.
    #[garde(skip)]
    pub stage: Option<Stage>,
    /// Only the user's own deals.
    #[garde(skip)]
    #[serde(default)]
    pub mine: bool,
}

#[derive(Debug, Deserialize, ToSchema, Validate)]
pub struct DealRef {
    /// The deal's id, from `search_deals`.
    #[garde(skip)]
    pub deal_id: Uuid,
}

#[derive(Debug, Deserialize, ToSchema, Validate)]
pub struct LogActivity {
    /// The deal's id, from `search_deals`.
    #[garde(skip)]
    pub deal_id: Uuid,
    /// `note`, `call`, `email`, `meeting`, or `task` for something to do later.
    #[garde(skip)]
    pub kind: ActivityKind,
    /// One line, e.g. "Call Dana about the revised quote".
    #[garde(length(chars, min = 1, max = 255))]
    pub subject: String,
    /// Details, if any.
    #[garde(length(chars, max = 5000))]
    pub body: Option<String>,
    /// For tasks: when it is due (RFC 3339, UTC).
    #[garde(skip)]
    pub due_at: Option<DateTime<Utc>>,
}

/// Gives a deal to another member of the team.
#[tool(needs_approval)]
#[can(Manage, Deal)]
async fn reassign_deal(&self, run: &AgentRun, args: ReassignDeal) -> Result<DealResource> {
    // ...
}

/// Moves a deal to another stage. `lost` needs a reason.
#[tool(needs_approval)]
async fn move_deal(&self, run: &AgentRun, args: MoveDealArgs) -> Result<DealResource> {
    let deal = run.authorize::<Deal, Move>(args.deal_id).await?;
    self.deals
        .move_to(
            run.uow(),
            run.user_id(),
            deal.into_inner(),
            MoveDeal { stage: Some(args.stage), position: None, lost_reason: args.lost_reason },
        )
        .await
}

{"error": {"statusCode": 403, "message": "Forbidden", "error": "Forbidden", "code": "forbidden"}}

use navree::access::CurrentAccess;
use navree::ai::{Ai, AgentReply};

#[derive(Debug, Deserialize, Validate, ToSchema)]
pub struct AskAssistant {
    #[garde(length(chars, min = 1, max = 4000))]
    pub message: String,
    /// Continue this conversation; leave out to start a new one.
    #[garde(skip)]
    pub conversation_id: Option<Uuid>,
}

#[injectable]
pub struct AssistantController {
    ai: Arc<Ai>,
}

#[controller("/assistant", version = "1")]
impl AssistantController {
    /// Sends a message to the deals assistant and returns its reply.
    #[post("/messages")]
    #[can(Read, Deal)]
    async fn ask(
        &self,
        CurrentAccess(access): CurrentAccess,
        Valid(Json(input)): Valid<Json<AskAssistant>>,
    ) -> Result<Json<AgentReply>> {
        let reply = self
            .ai
            .agent::<DealsAssistant>()
            .acting_as(access)
            .conversation(input.conversation_id)
            .prompt(input.message)
            .await?;
        Ok(Json(reply))
    }
}

POST /api/v1/assistant/messages HTTP/1.1
Authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...
Content-Type: application/json

{"message": "Which of my deals in negotiation have had no activity for two weeks?"}

{
  "conversation_id": "0199c3a4-1b2e-7c40-9d1e-5a6b7c8d9e0f",
  "text": "Two of your deals in negotiation have been quiet since October 9: Acme renewal ($12,000) and Globex expansion ($48,500). Want me to add a call task for each?",
  "steps": [
    {"tool": "search_deals", "arguments": {"stage": "negotiation", "mine": true}, "ok": true},
    {"tool": "deal_activity", "arguments": {"deal_id": "0199c1d2-6f7e-7a31-b5d4-2f0c8e9a1b23"}, "ok": true},
    {"tool": "deal_activity", "arguments": {"deal_id": "0199c1d4-0a11-7b52-8e3f-6c7d8e9f0a1b"}, "ok": true}
  ],
  "pending_approval": null,
  "usage": {"input_tokens": 6120, "output_tokens": 214, "cached_input_tokens": 3904}
}

m.schedule(|s| {
    s.command("ai:prune", &[]).daily_at("03:30").without_overlapping();
});

use navree::ai::Conversations;

let recent = conversations.list_for(user.id, PageQuery::default()).await?;  // Page<ConversationSummary>
let messages = conversations.messages(user.id, conversation_id).await?;     // Vec<StoredMessage>
conversations.rename(user.id, conversation_id, "Q4 pipeline").await?;
conversations.delete(user.id, conversation_id).await?;

{
  "conversation_id": "0199c3a4-1b2e-7c40-9d1e-5a6b7c8d9e0f",
  "text": "I'll move Acme renewal to Won. Shall I go ahead?",
  "steps": [{"tool": "search_deals", "arguments": {"query": "Acme renewal"}, "ok": true}],
  "pending_approval": {
    "call_id": "toolu_01JdX9wXyRk5b2T7",
    "tool": "move_deal",
    "arguments": {"deal_id": "0199c1d2-6f7e-7a31-b5d4-2f0c8e9a1b23", "stage": "won"}
  },
  "usage": {"input_tokens": 2810, "output_tokens": 96, "cached_input_tokens": 0}
}

#[derive(Debug, Deserialize, Validate, ToSchema)]
pub struct Decision {
    #[garde(skip)]
    pub conversation_id: Uuid,
    /// Why the user said no, passed on to the model.
    #[garde(length(chars, max = 500))]
    pub reason: Option<String>,
}

/// Runs a tool call the assistant asked to make, then continues.
#[post("/messages/{call_id}/approve")]
#[can(Read, Deal)]
async fn approve(
    &self,
    CurrentAccess(access): CurrentAccess,
    Path(call_id): Path<String>,
    Valid(Json(input)): Valid<Json<Decision>>,
) -> Result<Json<AgentReply>> {
    let reply = self
        .ai
        .agent::<DealsAssistant>()
        .acting_as(access)
        .conversation(Some(input.conversation_id))
        .approve(call_id) // runs the tool, then the model continues
        .await?;
    Ok(Json(reply))
}

// `reject` is the same with `.reject(call_id, input.reason)`: the model is
// told the user declined, and why.

#[post("/messages/stream")]
#[can(Read, Deal)]
async fn ask_stream(
    &self,
    CurrentAccess(access): CurrentAccess,
    Valid(Json(input)): Valid<Json<AskAssistant>>,
) -> Result<AgentStream> {
    Ok(self
        .ai
        .agent::<DealsAssistant>()
        .acting_as(access)
        .conversation(input.conversation_id)
        .stream(input.message)
        .await?)
}

event: tool_call
data: {"id":"toolu_01A","tool":"search_deals","arguments":{"stage":"negotiation","mine":true}}

event: tool_result
data: {"id":"toolu_01A","tool":"search_deals","ok":true}

event: text
data: {"text":"Two of your deals in negotiation"}

event: text
data: {"text":" have been quiet since October 9."}

event: done
data: {"model":"smart","provider":"anthropic","usage":{"input_tokens":6120,"output_tokens":214,"cached_input_tokens":0},"finish":"stop","conversation_id":"0199c3a4-1b2e-7c40-9d1e-5a6b7c8d9e0f"}

use navree::ai::Ai;
use navree::contracts::{BoxError, Job, JobContext};
use navree::mail::{Mailer, TemplateMail};
use serde::{Deserialize, Serialize};
use serde_json::json;
use uuid::Uuid;

use crate::assistant::DealsAssistant;

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct WeeklyPipelineReview {
    pub user_id: Uuid,
    pub email: String,
}

impl Job for WeeklyPipelineReview {
    const NAME: &'static str = "deals.weekly_review";

    async fn handle(self, ctx: JobContext) -> Result<(), BoxError> {
        let ai = ctx.app.get::<Ai>().await?;
        let review = ai
            .agent::<DealsAssistant>()
            .acting_as_user(self.user_id)
            .read_only()
            .prompt(
                "Review my open deals. List the three that most need attention this week, \
                 with one concrete next step each. Plain text, no tables.",
            )
            .await?;

        let mailer = ctx.app.get::<Mailer>().await?;
        mailer
            .send(
                &TemplateMail::new(self.email, "Your pipeline this week", "mail/weekly-review")
                    .context(json!({ "review": review.text })),
            )
            .await?;
        Ok(())
    }
}

use crm_testkit::{Role, TestCrm};
use navree::ai::{Ai, FakeReply};
use serde_json::json;

#[tokio::test]
async fn a_rep_cannot_move_another_reps_deal_through_the_assistant() {
    let crm = TestCrm::builder()
        .app(crm_api::app)
        .config("ai.fake", true)
        .build()
        .await;
    let ai = crm.get::<Ai>().await.fake().unwrap();

    let owner = crm.login_as(Role::Rep1).await;
    let deal = create_deal(&owner, "Acme renewal").await; // POSTs a company and a deal

    ai.respond_with([
        FakeReply::tool_call("move_deal", json!({ "deal_id": deal["id"], "stage": "proposal" })),
        FakeReply::text("I can't move that deal: it belongs to someone else."),
    ]);

    let other = crm.login_as(Role::Rep2).await;
    let reply = other
        .post("/api/v1/assistant/messages")
        .json(&json!({ "message": "Move Acme renewal to proposal" }))
        .send()
        .await
        .assert_status(200)
        .json_value();

    // `move_deal` needs approval: approve it, so the tool runs and is refused.
    let call_id = reply["pending_approval"]["call_id"].as_str().unwrap();
    other
        .post(&format!("/api/v1/assistant/messages/{call_id}/approve"))
        .json(&json!({ "conversation_id": reply["conversation_id"] }))
        .send()
        .await
        .assert_status(200)
        .assert_json_path("steps.0.ok", json!(false));

    let result = &ai.tool_results("move_deal")[0];
    assert_eq!(result["error"]["statusCode"], 403);
    owner
        .get(&format!("/api/v1/deals/{}", deal["id"].as_str().unwrap()))
        .send()
        .await
        .assert_json_path("stage", json!("lead"));
}