aion-server 0.29.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, 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";

/// 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 }),
                }]
            }
            ActivityEventKind::Progress {
                detail: ProgressDetail::Note { 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 == 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 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;