aion-server 0.31.0

//! Translating one turn's harness transcript into the session frames the
//! console reads and the store keeps.
//!
//! The adapter hands out neutral [`ActivityEvent`]s — the same shape a worker's
//! agent activity produces. This is the ONE place they become assistant frames,
//! so the socket and the durable transcript can never carry two different
//! renderings of the same event.
//!
//! # Nothing is dropped
//!
//! The ACP adapter's own rule is that every frame reaches the transcript, either
//! classified or verbatim. That rule survives this translation: an event with no
//! assistant frame of its own becomes [`AssistantSessionEvent::Raw`] carrying
//! the adapter's source label and the value untouched. A translation that
//! silently produced nothing would be the one place an operator's record could
//! lose a frame.
//!
//! # Two events that need memory
//!
//! ACP reports a tool RESULT without repeating the tool's name, and it reports a
//! permission decision in a frame after the request it decides. Both are joined
//! here, per turn, by [`TurnFrames`] — which is why this is a stateful
//! translator and not a free function.

use aion_core::{
    ActivityEvent, ActivityEventKind, AssistantCommand, AssistantConfigChoice,
    AssistantConfigOption, AssistantConfigValue, AssistantPermissionDecision,
    AssistantSessionEvent, AssistantToolCallStatus, MessageRole, ProgressDetail,
};
use serde_json::Value;
use std::collections::HashMap;

/// The adapter's source label for a permission REQUEST.
const PERMISSION_REQUEST_SOURCE: &str = "session/request_permission";
/// The adapter's source label for the DECISION that answered one.
const PERMISSION_DECISION_SOURCE: &str = "session/request_permission/decision";
/// The adapter's source label for the harness's command advertisement.
///
/// The adapter passes `available_commands_update` through verbatim because the
/// neutral activity vocabulary has no shape for it. It has one HERE — a session
/// is a conversation, and what an operator may ask for is part of what a
/// conversation is — so the frame is classified rather than left as a `raw` the
/// console would have to parse an ACP payload out of.
const AVAILABLE_COMMANDS_SOURCE: &str = "session/update/available_commands_update";

/// The adapter's source label for the harness's configuration-option
/// advertisement — the model picker's wire, among whatever else the agent
/// offers. Classified here for exactly the reason the command advertisement
/// is: what an operator may configure is part of what a conversation is.
const CONFIG_OPTIONS_SOURCE: &str = "session/update/config_option_update";

/// Per-turn translation state.
///
/// One of these lives for one turn, in the task that drains that turn's events,
/// so two concurrent sessions cannot see each other's tool names or pending
/// permission requests. (Two concurrent turns on ONE session cannot happen: the
/// adapter refuses a second open turn.)
pub(crate) struct TurnFrames {
    turn_id: String,
    /// `call_id -> tool name`, so a result can name the tool that produced it.
    tool_names: HashMap<String, String>,
    /// The permission request awaiting its decision, verbatim.
    pending_permission: Option<Value>,
}

impl TurnFrames {
    /// Start translating the turn `turn_id`.
    pub(crate) fn new(turn_id: impl Into<String>) -> Self {
        Self {
            turn_id: turn_id.into(),
            tool_names: HashMap::new(),
            pending_permission: None,
        }
    }

    /// Translate one adapter event into the frames it produces.
    ///
    /// Usually one; a permission REQUEST produces none on its own (it is held
    /// until the decision that answers it arrives) and the decision then
    /// produces the single joined frame.
    pub(crate) fn translate(&mut self, event: ActivityEvent) -> Vec<AssistantSessionEvent> {
        match event.kind {
            ActivityEventKind::Delta { text_fragment, .. } => {
                vec![AssistantSessionEvent::Delta {
                    turn_id: self.turn_id.clone(),
                    text: text_fragment,
                }]
            }
            // TWO events produce no frame, for two reasons that happen to
            // share an answer.
            //
            // A completed ASSISTANT message is the durable twin of the deltas
            // already streamed, so forwarding it too would double every answer
            // on the console AND in the transcript. The deltas ARE persisted
            // here, so replay shows exactly what live showed.
            //
            // A terminal STOP is reported by the turn's own result, which
            // carries the stop reason AND the final message; a second terminal
            // frame here would be a boundary the console had to reconcile.
            ActivityEventKind::Message {
                role: MessageRole::Assistant,
                ..
            }
            | ActivityEventKind::Stop { .. } => Vec::new(),
            ActivityEventKind::Message { role, text } => {
                vec![AssistantSessionEvent::Raw {
                    turn_id: Some(self.turn_id.clone()),
                    source: "message".to_owned(),
                    value: serde_json::json!({ "role": role, "text": text }),
                }]
            }
            // Reasoning is what the wire SAYS is reasoning. It used to be
            // read off a free-form `Note`, because that was where the ACP
            // adapter had to put a thought chunk; when the adapter moved to
            // `Thinking` this arm still named `Note`, and every thought fell
            // through the catch-all below into a raw frame the console draws
            // as nothing — found in review of aion#224. A `Note` is progress
            // text, not a thought, and takes the raw arm on purpose.
            ActivityEventKind::Progress {
                detail: ProgressDetail::Thinking { text, .. },
            } => vec![AssistantSessionEvent::Thought {
                turn_id: self.turn_id.clone(),
                text,
            }],
            ActivityEventKind::Progress { detail } => {
                vec![AssistantSessionEvent::Raw {
                    turn_id: Some(self.turn_id.clone()),
                    source: "progress".to_owned(),
                    value: serde_json::to_value(detail).unwrap_or(Value::Null),
                }]
            }
            ActivityEventKind::ToolCall {
                tool,
                call_id,
                input,
            } => {
                self.tool_names.insert(call_id.clone(), tool.clone());
                vec![AssistantSessionEvent::ToolCall {
                    turn_id: self.turn_id.clone(),
                    call_id,
                    name: tool,
                    status: AssistantToolCallStatus::Started,
                    input: Some(input),
                    output: None,
                }]
            }
            ActivityEventKind::ToolResult {
                call_id,
                output,
                is_error,
            } => {
                // The name is remembered from the call. A result with no
                // remembered call is a conformance problem on the agent's side,
                // and the frame says so by name rather than inventing a tool.
                let name = self
                    .tool_names
                    .get(&call_id)
                    .cloned()
                    .unwrap_or_else(|| UNMATCHED_TOOL.to_owned());
                vec![AssistantSessionEvent::ToolCall {
                    turn_id: self.turn_id.clone(),
                    call_id,
                    name,
                    status: if is_error {
                        AssistantToolCallStatus::Failed
                    } else {
                        AssistantToolCallStatus::Completed
                    },
                    input: None,
                    output: Some(output),
                }]
            }
            ActivityEventKind::Raw { source, value } => self.translate_raw(&source, value),
        }
    }

    /// The raw sources that carry meaning, and the passthrough for the rest.
    fn translate_raw(&mut self, source: &str, value: Value) -> Vec<AssistantSessionEvent> {
        if source == AVAILABLE_COMMANDS_SOURCE {
            return match available_commands(&value) {
                Some(commands) => vec![AssistantSessionEvent::AvailableCommands { commands }],
                // The frame arrived and could not be read as a command list.
                // Passed through verbatim rather than dropped or reported as an
                // EMPTY advertisement: an empty list is a withdrawal, and
                // withdrawing every command because a payload changed shape
                // would take a working control off an operator's surface.
                None => vec![AssistantSessionEvent::Raw {
                    turn_id: Some(self.turn_id.clone()),
                    source: source.to_owned(),
                    value,
                }],
            };
        }
        if source == CONFIG_OPTIONS_SOURCE {
            return match config_options(&value) {
                Some(options) => vec![AssistantSessionEvent::ConfigOptions { options }],
                // Same rule as an unreadable command advertisement, for the
                // same reason: an empty list is a withdrawal, and withdrawing
                // every option because a payload changed shape would take the
                // model picker off an operator's surface.
                None => vec![AssistantSessionEvent::Raw {
                    turn_id: Some(self.turn_id.clone()),
                    source: source.to_owned(),
                    value,
                }],
            };
        }
        if source == PERMISSION_REQUEST_SOURCE {
            // Held, not emitted: the frame the console renders states what was
            // asked AND what was answered, and the answer has not happened yet.
            self.pending_permission = Some(value);
            return Vec::new();
        }
        if source == PERMISSION_DECISION_SOURCE {
            let request = self.pending_permission.take().unwrap_or(Value::Null);
            let Some(decided) = decision_of(&value) else {
                // A CANCELLED outcome is neither allowed nor denied: nothing was
                // decided about the call, and reporting it as a denial would put
                // a decision on the record that was never taken. It reaches the
                // transcript verbatim instead, with the ask beside it, so the
                // record is complete and the classified frame stays honest about
                // meaning exactly one of two things.
                return vec![AssistantSessionEvent::Raw {
                    turn_id: Some(self.turn_id.clone()),
                    source: source.to_owned(),
                    value: serde_json::json!({ "request": request, "decision": value }),
                }];
            };
            return vec![AssistantSessionEvent::PermissionAsk {
                turn_id: self.turn_id.clone(),
                request,
                decided,
            }];
        }
        vec![AssistantSessionEvent::Raw {
            turn_id: Some(self.turn_id.clone()),
            source: source.to_owned(),
            value,
        }]
    }

    /// The permission request that was never answered, if the turn ended with
    /// one held.
    ///
    /// A turn that dies mid-decision leaves a request recorded by the adapter
    /// and no decision. Emitting it here — with `decided: cancelled`, which is
    /// what a client that stops MUST answer — keeps the ask on the record
    /// instead of dropping it with the translator.
    pub(crate) fn flush(&mut self) -> Vec<AssistantSessionEvent> {
        self.pending_permission
            .take()
            .map(|request| AssistantSessionEvent::Raw {
                turn_id: Some(self.turn_id.clone()),
                source: format!("{PERMISSION_REQUEST_SOURCE}/unanswered"),
                value: request,
            })
            .into_iter()
            .collect()
    }
}

/// The name reported for a tool result whose call was never seen.
pub(crate) const UNMATCHED_TOOL: &str = "<unmatched tool call>";

/// Read ACP's `available_commands_update` payload into the published shape.
///
/// `None` when the payload is not a command list at all — which is different
/// from a list with nothing in it, and must stay different: an empty list is the
/// agent WITHDRAWING every command, and an unreadable payload is this server not
/// knowing what the agent said.
///
/// A single entry that is malformed is dropped with a log line rather than
/// taking the whole advertisement down: the other commands the agent named are
/// still offerable, and a command this server could not read is one it could not
/// have offered anyway.
fn available_commands(value: &Value) -> Option<Vec<AssistantCommand>> {
    let listed = value.get("availableCommands")?.as_array()?;
    let mut commands = Vec::with_capacity(listed.len());
    for entry in listed {
        let (Some(name), Some(description)) = (
            entry.get("name").and_then(Value::as_str),
            entry.get("description").and_then(Value::as_str),
        ) else {
            tracing::warn!(
                entry = %entry,
                "an assistant harness advertised a command with no name or no description; it is \
                 not offered, and the rest of the advertisement stands"
            );
            continue;
        };
        commands.push(AssistantCommand {
            name: name.to_owned(),
            description: description.to_owned(),
            // ACP's only input form is `unstructured`, whose whole published
            // content is the hint. A future form with more in it would land
            // here as `None` and the command would still be offerable.
            input_hint: entry
                .get("input")
                .and_then(|input| input.get("hint"))
                .and_then(Value::as_str)
                .map(ToOwned::to_owned),
        });
    }
    Some(commands)
}

/// Read ACP's `config_option_update` payload (or the identically-shaped
/// `configOptions` list a `session/new`, `session/load` or
/// `session/set_config_option` response carries) into the published shape.
///
/// `None` when the payload carries no readable option list at all — which is
/// different from a list with nothing in it, for the reason
/// [`available_commands`] states. A single entry that is malformed, or whose
/// kind this server has no shape for, is skipped with a log line: the other
/// options the agent named are still offerable.
pub(crate) fn config_options(value: &Value) -> Option<Vec<AssistantConfigOption>> {
    let listed = value.get("configOptions")?.as_array()?;
    let mut options = Vec::with_capacity(listed.len());
    for entry in listed {
        let (Some(id), Some(name)) = (
            entry.get("id").and_then(Value::as_str),
            entry.get("name").and_then(Value::as_str),
        ) else {
            tracing::warn!(
                entry = %entry,
                "an assistant harness advertised a configuration option with no id or no name; \
                 it is not offered, and the rest of the advertisement stands"
            );
            continue;
        };
        let Some(value_read) = config_value(entry) else {
            tracing::warn!(
                option = id,
                kind = entry
                    .get("type")
                    .and_then(|value| value.as_str())
                    .unwrap_or("<absent>"),
                "an assistant harness advertised a configuration option of a kind this server \
                 has no shape for; it is not offered, and the rest of the advertisement stands"
            );
            continue;
        };
        options.push(AssistantConfigOption {
            id: id.to_owned(),
            name: name.to_owned(),
            description: entry
                .get("description")
                .and_then(Value::as_str)
                .map(ToOwned::to_owned),
            category: entry
                .get("category")
                .and_then(Value::as_str)
                .map(ToOwned::to_owned),
            value: value_read,
        });
    }
    Some(options)
}

/// One option's type-specific half, or `None` for a kind with no shape here.
fn config_value(entry: &Value) -> Option<AssistantConfigValue> {
    match entry.get("type").and_then(Value::as_str) {
        Some("select") => {
            let current = entry
                .get("currentValue")
                .and_then(Value::as_str)?
                .to_owned();
            let listed = entry.get("options").and_then(Value::as_array)?;
            let mut choices = Vec::with_capacity(listed.len());
            for element in listed {
                // ACP's select options are either a flat list of values or a
                // list of GROUPS each holding values; both arrive here. The
                // group's label rides along on each choice, so the flattening
                // loses nothing a surface may want to draw.
                if let Some(grouped) = element.get("options").and_then(Value::as_array) {
                    let group = element.get("name").and_then(Value::as_str);
                    for inner in grouped {
                        push_choice(&mut choices, inner, group);
                    }
                } else {
                    push_choice(&mut choices, element, None);
                }
            }
            Some(AssistantConfigValue::Select { choices, current })
        }
        Some("boolean") => Some(AssistantConfigValue::Toggle {
            current: entry.get("currentValue").and_then(Value::as_bool)?,
        }),
        _ => None,
    }
}

/// Read one select choice, skipping a malformed one with a log line.
fn push_choice(choices: &mut Vec<AssistantConfigChoice>, element: &Value, group: Option<&str>) {
    let (Some(id), Some(name)) = (
        element.get("value").and_then(Value::as_str),
        element.get("name").and_then(Value::as_str),
    ) else {
        tracing::warn!(
            entry = %element,
            "an assistant harness advertised a select choice with no value or no name; it is \
             not offered, and the rest of the advertisement stands"
        );
        return;
    };
    choices.push(AssistantConfigChoice {
        id: id.to_owned(),
        name: name.to_owned(),
        description: element
            .get("description")
            .and_then(Value::as_str)
            .map(ToOwned::to_owned),
        group: group.map(ToOwned::to_owned),
    });
}

/// Read the decision out of the adapter's decision frame, or `None` when the
/// frame records that nothing was decided.
///
/// The frame's `outcome` is the ACP outcome verbatim; its `outcome.outcome`
/// discriminator is `selected` for a chosen option and `cancelled` for a turn
/// that had already been cancelled, and the adapter's policy selects a reject
/// option for a deny. The policy label beside it distinguishes the two
/// `selected` cases.
///
/// `None` is the cancelled case and nothing else: the classified frame carries
/// exactly two answers, so a third fact must not be squeezed into one of them.
fn decision_of(value: &Value) -> Option<AssistantPermissionDecision> {
    if value["outcome"]["outcome"] == "cancelled" {
        return None;
    }
    Some(match value["policy"].as_str() {
        Some("allow_once") => AssistantPermissionDecision::AllowOnce,
        // `deny`, and anything a future policy label could be: the SAFE reading
        // of an unrecognised policy is that it did not allow the call.
        _ => AssistantPermissionDecision::Deny,
    })
}

#[cfg(test)]
#[path = "frames_tests.rs"]
mod tests;