basis 0.21.0

//! Handing a task to a remote agent, once the answer is in (ADR-0029).
//!
//! The agent is reached through the [`Peer`] its resolver handed back
//! (ADR-0038): the peer runs the exchange and frames the agent's words; what
//! is basis's is everything the model would otherwise have had to carry —
//! which conversation a task belongs to, and which task is still outstanding
//! (ADR-0029 §5). So every guidance line here is written in `spawn`'s
//! grammar, and none of them mentions an id.
//!
//! [`Peer`]: crate::agents::Peer

use std::{
    path::{Path, PathBuf},
    sync::Arc,
    time::SystemTime,
};

use mentra::{
    agent::DeferredItem,
    runtime::RunOptions,
    tool::{ParallelToolContext, ToolContext, ToolResult},
};

use super::{
    background::{Children, Kind, Settled, TaskScope},
    describe::listed,
    execute::started,
    files, no_remote_agents,
};
use crate::agents::{
    Attachment, ExchangeBounds, Failure, Message, Outcome, PeerError, Request, Resolved, Route,
    Started, StillWorking, Unresolved,
    remote::{Conversation, Outstanding, Remembered, RemoteAgents, Waiting},
};

/// Whether `@<name>` can go anywhere, asked in the preview — before the
/// approver, and without waiting on the network (ADR-0037 §4).
///
/// A known agent is let through whether or not it is reachable now:
/// resolving it is the first thing [`delegate`] does (ADR-0031 §2), and
/// refusing it here would mean an agent that came up is never reached
/// without a reload. So is a name a resolver cannot classify without the
/// network. One tried too recently is refused, with how long until it is
/// worth asking again.
pub(super) fn authorize(agents: &RemoteAgents, name: &str) -> Result<(), String> {
    match agents.route(name) {
        Route::Known | Route::Maybe => Ok(()),
        Route::Backoff(wait) => Err(backing_off(name, wait)),
        Route::Refused(reason) => Err(refused(name, &reason)),
        Route::Unknown => Err(unknown(agents, name)),
    }
}

/// The refusal for an agent its resolver refused, in the resolver's words,
/// then what to do instead (ADR-0031 §5).
fn refused(name: &str, reason: &str) -> String {
    format!(
        "{reason}, so there is nothing to hand `@{name}` to; write the task with no `@` to hand \
         it to a subagent here"
    )
}

/// The refusal for a name no resolver of the calling agent's workspace knows.
pub(super) fn unknown(agents: &RemoteAgents, name: &str) -> String {
    let configured = agents.names();
    if configured.is_empty() {
        return no_remote_agents(name);
    }
    format!(
        "there is no remote agent named `{name}`; the ones this workspace configures are {}, \
         and `@@` begins a task with a literal `@`",
        listed(&configured)
    )
}

/// The refusal for an unreachable agent tried too recently to try again.
fn backing_off(name: &str, wait: std::time::Duration) -> String {
    format!(
        "the remote agent `{name}` could not be reached when it was last tried; `@{name}` tries \
         it again in {}s",
        seconds(wait)
    )
}

/// Whole seconds, never zero: "again in 0s" reads as "now", which it is not.
fn seconds(wait: std::time::Duration) -> u64 {
    wait.as_secs_f64().ceil().max(1.0) as u64
}

/// What the model reads when there is no peer to hand the task to.
fn unresolved(agents: &RemoteAgents, name: &str, why: Unresolved) -> String {
    match why {
        Unresolved::Backoff(wait) => backing_off(name, wait),
        Unresolved::Unreachable {
            error,
            retry_in: Some(retry_in),
        } => format!(
            "the remote agent `{name}` could not be reached: {error}. `@{name}` tries it again \
             in {}s",
            seconds(retry_in)
        ),
        Unresolved::Unreachable {
            error,
            retry_in: None,
        } => format!("the remote agent `{name}` could not be reached: {error}"),
        Unresolved::Refused(reason) => refused(name, &reason),
        Unresolved::Unknown => unknown(agents, name),
    }
}

/// Hands `body` to the remote agent `name`, or — with an empty body — waits
/// again on its outstanding task, and reports what came back.
///
/// `remembered` says where the calling agent's record is kept: on disk by
/// agent for one a workspace minted or resumed, under a host's key for one
/// bound to it, in memory for a delegated child (`crate::agents::remote`
/// has the rule).
/// What one call hands a remote agent: the task in words, the parameters
/// beside it (ADR-0040), and the files with it (ADR-0041). An empty body is a
/// wait, which carries neither.
pub(super) struct Task {
    pub(super) body: String,
    pub(super) metadata: Option<serde_json::Map<String, serde_json::Value>>,
    pub(super) files: Vec<Attachment>,
}

/// This side of an exchange: who is asking, under what bounds, where
/// progress goes, and how files that come back are written. The blocking
/// path answers all four from the tool context; a background task (ADR-0042)
/// has no context, so its progress goes nowhere and its files are written
/// straight into the workspace inbox.
struct Side<'a> {
    caller: String,
    options: RunOptions,
    progress: &'a (dyn Fn(String) + Sync),
    root: PathBuf,
    write: Write<'a>,
}

/// How a returned file is written: the file tools' policy, or the inbox.
type Write<'a> = &'a (dyn Fn(&str, &[u8]) -> Result<PathBuf, String> + Sync);

/// Hands `task` to the remote agent `name`, or — with an empty body — waits
/// again on its outstanding task, and reports what came back.
///
/// `remembered` says where the calling agent's record is kept: on disk by
/// agent for one a workspace minted or resumed, under a host's key for one
/// bound to it, in memory for a delegated child (`crate::agents::remote`
/// has the rule).
pub(super) async fn delegate(
    ctx: ToolContext<'_>,
    agents: &RemoteAgents,
    name: &str,
    task: Task,
    remembered: &Remembered,
) -> ToolResult {
    let caller = ctx.agent_id.clone();
    // The run's own bounds — cancellation, stop, deadline — taken before the
    // context is handed over for progress, which is the one thing the
    // exchange needs from it while it waits. The deadline also travels: the
    // remote agent is told when this side stops waiting.
    let options = ctx.child_run_options();
    let ctx: ParallelToolContext = ctx.into();
    let root = ctx
        .resolve_working_directory(None)
        .unwrap_or_else(|_| ctx.working_directory().to_path_buf());
    let progress = |text: String| ctx.emit_progress(text);
    let write = |path: &str, bytes: &[u8]| ctx.write_file_bytes(path, bytes);
    run(
        agents,
        name,
        task,
        remembered,
        Side {
            caller,
            options,
            progress: &progress,
            root,
            write: &write,
        },
    )
    .await
}

/// Hands `task` to the remote agent `name` on a task of its own and returns
/// at once (ADR-0042). The handle is the agent's name: one task per remote
/// agent at a time is already the rule (ADR-0029 §5), so the name is the
/// conversation, and `@<name>` alone collects it as it always did.
///
/// What comes back is left for the parent's next round as a plain note, the
/// same words the blocking call would have returned, since a remote task
/// writes no delegation entry on either path.
pub(super) fn delegate_in_background(
    ctx: &ToolContext<'_>,
    scope: Option<&TaskScope>,
    children: Children,
    agents: Arc<RemoteAgents>,
    name: &str,
    task: Task,
    remembered: Remembered,
) -> ToolResult {
    let caller = ctx.agent_id.clone();
    let options = ctx.child_run_options();
    let root = ctx.resolve_working_directory(None)?;
    let agent = name.to_string();
    children.start(
        scope,
        name.to_string(),
        Some(name.to_string()),
        Kind::Remote,
        ctx.deferred(),
        async move {
            let progress = |_: String| {};
            let inbox = root.clone();
            let write = move |path: &str, bytes: &[u8]| write_under(&inbox, path, bytes);
            let outcome = run(
                &agents,
                &agent,
                task,
                &remembered,
                Side {
                    caller,
                    options,
                    progress: &progress,
                    root,
                    write: &write,
                },
            )
            .await;
            let (told, text) = match &outcome {
                Ok(text) => ("finished", text.clone()),
                Err(text) => ("failed", text.clone()),
            };
            Settled {
                outcome,
                items: vec![DeferredItem::Context(format!(
                    "<delegation-result agent=\"{agent}\" status=\"{told}\">\n{text}\n\
                     </delegation-result>"
                ))],
            }
        },
    );
    Ok(started(Kind::Remote, name, Some(name), false))
}

/// Writes `bytes` at `path` under `root` — the inbox path `files::save`
/// chose — for a background task that has no file-tool context to write
/// through. The path is basis's own (`.basis/inbox/…`), inside the
/// workspace by construction.
fn write_under(root: &Path, path: &str, bytes: &[u8]) -> Result<PathBuf, String> {
    let full = root.join(path);
    if let Some(parent) = full.parent() {
        std::fs::create_dir_all(parent).map_err(|error| error.to_string())?;
    }
    std::fs::write(&full, bytes).map_err(|error| error.to_string())?;
    Ok(full)
}

async fn run(
    agents: &RemoteAgents,
    name: &str,
    task: Task,
    remembered: &Remembered,
    side: Side<'_>,
) -> ToolResult {
    let Task {
        body,
        metadata,
        files,
    } = task;
    // Held to the end of this call: a reload that swaps the set meanwhile
    // changes what the *next* `@<agent>` reaches, never this exchange
    // (ADR-0031 §6). An unreachable agent due a retry is connected here.
    let Resolved {
        peer, id, endpoint, ..
    } = agents
        .resolve(name)
        .await
        .map_err(|why| unresolved(agents, name, why))?;
    // The conversation is the agent's, not the name's (ADR-0037 §6): kept
    // under the id its resolver gave it, so a short handle and its full form
    // are one conversation, and a label handed to another agent does not
    // inherit one.
    let id = id.as_str();
    let caller = side.caller.as_str();
    let options = side.options;
    let hop = Hop {
        hop: agents.hop().saturating_add(1),
        deadline: options.deadline,
    };
    let before = agents.conversation(caller, id, remembered);
    let request = request(name, &body, &before, hop)?;
    let request = match request {
        Request::Send(message) => Request::Send(message.with_metadata(metadata).with_files(files)),
        wait => wait,
    };

    let progress = |text: String| (side.progress)(text);
    let save = |next: Conversation| {
        // Reported, not raised: the exchange it describes happened, and a
        // failed tool call would invite the model to do the work again.
        if let Err(error) = agents.record(caller, id, next, remembered) {
            (side.progress)(format!(
                "the conversation with `{name}` was not saved: {error}"
            ));
        }
    };
    // Recorded before the peer waits, which is the whole point of being
    // told: a process that dies while this waits comes back knowing the task
    // exists, and `@<agent>` alone picks it up instead of the model handing
    // the work off a second time.
    let started = |started: Started| {
        save(Conversation {
            context_id: started.context_id.or_else(|| before.context_id.clone()),
            outstanding: Some(Outstanding {
                task_id: started.task_id,
                waiting_for: Waiting::Agent,
                endpoint: Some(endpoint.clone()),
            }),
        });
    };

    let outcome = peer
        .exchange(
            request,
            &ExchangeBounds {
                options: &options,
                progress: &progress,
                started: &started,
            },
        )
        .await;

    // The files come off the outcome before it is told: saving them is this
    // side's work, and what the model reads about them goes after the
    // agent's words, outside its frame (ADR-0041 §4).
    let (outcome, returned) = match outcome {
        Ok(mut outcome) => {
            let returned = take_files(&mut outcome);
            (Ok(outcome), returned)
        }
        Err(error) => (Err(error), None),
    };
    let (next, result) = settle(name, before, outcome, &endpoint);
    save(next);
    let Some((folder, files)) = returned else {
        return result;
    };
    let saved = files::save(&side.root, &folder, files, |path, bytes| {
        (side.write)(path, bytes)
    });
    match result {
        Ok(text) => Ok(said(text, saved)),
        Err(text) => Err(said(text, saved)),
    }
}

/// The files an outcome carries, taken off it, with the folder they are saved
/// in: the task's id, or for a reply that started none, one of its own.
/// `None` when it carries none.
fn take_files(outcome: &mut Outcome) -> Option<(String, Vec<Attachment>)> {
    let (task_id, files) = match outcome {
        Outcome::Replied { files, .. } => (None, files),
        Outcome::Finished { task_id, files, .. }
        | Outcome::Working { task_id, files, .. }
        | Outcome::NeedsInput { task_id, files, .. }
        | Outcome::NeedsAuthorization { task_id, files, .. }
        | Outcome::Failed { task_id, files, .. } => (Some(task_id.clone()), files),
        // `Outcome` is non-exhaustive; a newer outcome's files are not known.
        #[allow(unreachable_patterns)]
        _ => return None,
    };
    if files.is_empty() {
        return None;
    }
    let folder = task_id.unwrap_or_else(|| {
        let now = SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default();
        format!("reply-{}", now.as_millis())
    });
    Some((folder, std::mem::take(files)))
}

/// Where one outbound message stands in a chain of agents: `hop` is this
/// message's — the runtime's own
/// plus one, so a top-level caller sends 1 — and `deadline` is when this
/// side stops waiting, which the run's options already hold as the earlier
/// of its own and any it inherited.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct Hop {
    hop: u32,
    deadline: Option<SystemTime>,
}

/// What this call sends, given where the conversation stands — or why it
/// sends nothing.
fn request(
    name: &str,
    body: &str,
    conversation: &Conversation,
    hop: Hop,
) -> Result<Request, String> {
    match (&conversation.outstanding, body.is_empty()) {
        (Some(outstanding), true) => Ok(Request::wait(&outstanding.task_id)),
        (None, true) => Err(format!(
            "`{name}` has nothing outstanding to wait on; write `@{name} <task>` to hand it one"
        )),
        (
            Some(Outstanding {
                waiting_for: Waiting::Agent,
                ..
            }),
            false,
        ) => Err(format!(
            "`{name}` is still working on the last task you handed it, and takes one task at a \
             time; write `@{name}` alone to keep waiting for that one first"
        )),
        (Some(outstanding), false) => Ok(Request::Send(
            Message::new(body)
                .in_context(conversation.context_id.clone())
                .answering(Some(outstanding.task_id.clone()))
                .at_hop(hop.hop, hop.deadline),
        )),
        (None, false) => Ok(Request::Send(
            Message::new(body)
                .in_context(conversation.context_id.clone())
                .at_hop(hop.hop, hop.deadline),
        )),
    }
}

/// Where the conversation stands after this exchange, and what the model
/// reads about it. `endpoint` is the one this exchange was sent to, recorded
/// with any task left outstanding.
fn settle(
    name: &str,
    before: Conversation,
    outcome: Result<Outcome, PeerError>,
    endpoint: &str,
) -> (Conversation, ToolResult) {
    let outstanding = |task_id: String, waiting_for: Waiting| {
        Some(Outstanding {
            task_id,
            waiting_for,
            endpoint: Some(endpoint.to_string()),
        })
    };
    let settled = |context_id: Option<String>| Conversation {
        context_id: context_id.or_else(|| before.context_id.clone()),
        outstanding: None,
    };
    match outcome {
        Ok(Outcome::Replied {
            context_id, text, ..
        }) => (
            settled(context_id),
            Ok(said(format!("Remote agent `{name}` replied:"), text)),
        ),
        Ok(Outcome::Finished {
            context_id, text, ..
        }) => (
            settled(context_id),
            Ok(said(format!("Remote agent `{name}` replied:"), text)),
        ),
        Ok(Outcome::NeedsInput {
            task_id,
            context_id,
            text,
            ..
        }) => (
            Conversation {
                context_id: context_id.or_else(|| before.context_id.clone()),
                outstanding: outstanding(task_id, Waiting::Input),
            },
            Ok(said(
                format!(
                    "Remote agent `{name}` needs more before it can finish. Answer with \
                     `@{name} <your answer>`."
                ),
                text,
            )),
        ),
        Ok(Outcome::NeedsAuthorization {
            context_id, text, ..
        }) => (
            settled(context_id),
            Err(said(
                format!(
                    "Remote agent `{name}` needs authorization this conversation cannot give \
                     it, so the task stopped there."
                ),
                text,
            )),
        ),
        Ok(Outcome::Failed {
            context_id,
            text,
            how,
            ..
        }) => (
            settled(context_id),
            Err(said(
                format!(
                    "Remote agent `{name}` {}.",
                    match how {
                        Failure::Rejected => "rejected the task",
                        Failure::Canceled => "reports the task was canceled",
                        _ => "reports the task failed",
                    }
                ),
                text,
            )),
        ),
        Ok(Outcome::Working {
            task_id,
            context_id,
            text,
            why,
            ..
        }) => {
            let why = match why {
                StillWorking::WaitBound(waited) => {
                    format!("did not finish within {}s", waited.as_secs().max(1))
                }
                _ => "was still working when this run began to stop".to_string(),
            };
            (
                Conversation {
                    context_id: context_id.or_else(|| before.context_id.clone()),
                    outstanding: outstanding(task_id, Waiting::Agent),
                },
                Ok(said(
                    format!(
                        "Remote agent `{name}` {why}; it keeps working remotely. Write `@{name}` \
                         alone to keep waiting for it, and do not hand it the task again."
                    ),
                    text,
                )),
            )
        }
        // The agent moved since this task was handed over, and the new
        // endpoint could not pick it up — whatever it answered: one that does
        // not know a task need not say so (an HTTP 404, an invalid-params
        // refusal). Waiting again would fail the same way for good, so the
        // record ends here, and nothing is sent again.
        Err(error) if moved(&before, endpoint) && cause(&error).is_some() => (
            Conversation::default(),
            Err(format!(
                "Remote agent `{name}` has moved to a new address since it was handed the task it \
                 was working on, and the new address could not pick that task up ({}), so it is \
                 forgotten and the conversation with it starts over. Hand it the task again with \
                 `@{name} <task>` if it still needs doing.",
                cause(&error).expect("checked by the guard")
            )),
        ),
        Err(PeerError::UnknownTask) if before.outstanding.is_some() => (
            Conversation::default(),
            Err(format!(
                "Remote agent `{name}` no longer knows the task it was working on — it may have \
                 moved to a new address — so that task is forgotten and the conversation with \
                 it starts over. Hand it the task again with `@{name} <task>` if it still needs \
                 doing."
            )),
        ),
        Err(PeerError::UnknownTask) => (
            before,
            Err(format!(
                "Handing the task to remote agent `{name}` failed: it does not know the task \
                 this answers"
            )),
        ),
        // Nothing came back to wait on, whatever the cause says — even "it
        // may already be working on it", which forbids an automatic resend,
        // not a deliberate one. The model is told what it can do, or it
        // reads that as "wait", and a wait finds nothing outstanding.
        Err(PeerError::Send(error)) => (
            before,
            Err(format!(
                "Handing the task to remote agent `{name}` failed: {error}. No task came back, \
                 so there is nothing to wait on. If the task is safe to repeat, such as a \
                 question, hand it over again with `@{name} <task>`; if it is not, say that it \
                 may have been received."
            )),
        ),
        Err(PeerError::Check { task_id, error }) => (
            Conversation {
                outstanding: outstanding(task_id, Waiting::Agent),
                ..before
            },
            Err(format!(
                "Checking on remote agent `{name}` failed: {error}. The task may still be \
                 running; write `@{name}` alone to check on it again."
            )),
        ),
        Err(PeerError::Abandoned { canceled, .. }) => (
            Conversation {
                outstanding: None,
                ..before
            },
            Err(match canceled {
                Ok(()) => format!(
                    "The run was cancelled while remote agent `{name}` was working, and the task \
                     was asked to cancel."
                ),
                Err(error) => format!(
                    "The run was cancelled while remote agent `{name}` was working, and asking \
                     the task to cancel failed: {error}"
                ),
            }),
        ),
        Err(error) => (
            before,
            Err(format!(
                "Remote agent `{name}` could not be reached: {error}"
            )),
        ),
    }
}

/// Whether the outstanding task in `conversation` was handed to an endpoint
/// other than `endpoint`. A record that predates the field cannot say, and is
/// taken not to have moved.
fn moved(conversation: &Conversation, endpoint: &str) -> bool {
    conversation
        .outstanding
        .as_ref()
        .and_then(|outstanding| outstanding.endpoint.as_deref())
        .is_some_and(|recorded| recorded != endpoint)
}

/// The words behind a request or a check that failed — which name no task
/// id, unlike the error around it. `None` for an abandoned or unrecognized
/// exchange, which is not evidence of a move.
fn cause(error: &PeerError) -> Option<String> {
    match error {
        PeerError::Send(error) | PeerError::Check { error, .. } => Some(error.clone()),
        PeerError::UnknownTask => Some(error.to_string()),
        _ => None,
    }
}

/// A headline in basis's words, then the remote agent's own, already framed
/// as its output by the peer.
fn said(headline: String, output: String) -> String {
    if output.is_empty() {
        return headline;
    }
    format!("{headline}\n\n{output}")
}

#[cfg(test)]
mod tests;