aion-server 0.31.0

//! The declared command's output, into the transcript the server already
//! serves.
//!
//! A declared body executes IN the server process, so there is no worker drain
//! between the command and the sequencer: the activity's transcript seam is
//! drained here and published straight through
//! [`ActivityEventPublisher`](crate::activity_publisher::ActivityEventPublisher),
//! which is the same durable append, retention bound, and live fan-out an agent
//! step's transcript goes through after crossing the wire. Downstream — the
//! cursor reads, the WS tail, the ops console — cannot tell the two apart, and
//! is not meant to.
//!
//! # Losing a transcript never fails a command
//!
//! Publishing is observability, and the command's replay-authoritative result is
//! its own. A store that refuses events costs a log line and those events, never
//! the activity: the refusal is named once per refused BATCH — never once per
//! line of output — and the exact total is reported when the command ends. That
//! is the same coalescing discipline the worker-side drain uses, for the same
//! reason: a failing sink must not turn one line of output into one log line per
//! line of output.
//!
//! # One commit per batch, not per line
//!
//! A verbose command emits one transcript event per output line, and each
//! durable commit re-persists its whole containing storage leaf. This drain
//! therefore runs through the shared coalescing loop
//! ([`ActivityEventPublisher::drain`]) under the operator's
//! `observability.max_batch_events` / `max_batch_hold_ms` policy, exactly like
//! the liminal observability tap: a thousand-line command costs commits
//! proportional to its batches, not to its lines.

use aion_core::ActivityEvent;
use tokio::sync::mpsc;

use crate::activity_publisher::ActivityEventPublisher;

/// The `operation` label this drain's refusals are logged under.
const OPERATION: &str = "declared_command_dispatch";

/// Drain `events` into `publisher` until the activity's transcript seam closes.
///
/// Returns when the sender is dropped — which the declared-command executor does
/// as soon as its command has finished — so the caller can await this and know
/// every observed line has been offered to the sequencer.
pub(crate) async fn publish_declared_transcript(
    publisher: ActivityEventPublisher,
    mut events: mpsc::UnboundedReceiver<ActivityEvent>,
) {
    let dropped = publisher.drain(&mut events, OPERATION).await;
    if dropped > 0 {
        tracing::warn!(
            dropped,
            operation = OPERATION,
            "declared command transcript: output lines were not retained"
        );
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use aion_core::{ActivityEvent, ActivityEventKind, ActivityId, MessageRole, RunId, WorkflowId};
    use aion_store::{ActivityStreamKey, InMemoryObservabilityStore};
    use uuid::Uuid;

    use super::publish_declared_transcript;
    use crate::activity_publisher::ActivityEventPublisher;

    /// What a test returns. Every fallible step is carried rather than
    /// unwrapped, because the workspace denies panicking accessors in test code
    /// as firmly as in library code.
    type TestResult = Result<(), Box<dyn std::error::Error>>;

    fn line_event(worker_seq: u64, text: &str) -> ActivityEvent {
        ActivityEvent {
            workflow_id: WorkflowId::new(Uuid::from_u128(7)),
            run_id: RunId::new(Uuid::from_u128(0x70)),
            activity_id: ActivityId::from_sequence_position(2),
            attempt: 1,
            agent_id: Uuid::nil(),
            agent_role: "command stdout".to_owned(),
            emitted_at: chrono::Utc::now(),
            worker_seq,
            store_seq: None,
            ephemeral: false,
            kind: ActivityEventKind::Message {
                role: MessageRole::Tool,
                text: text.to_owned(),
            },
        }
    }

    /// Every drained line is sequenced into the durable transcript keyspace, in
    /// order, under the stream key the event carries.
    #[tokio::test]
    async fn drained_lines_are_sequenced_into_the_durable_transcript() -> TestResult {
        let store = Arc::new(InMemoryObservabilityStore::default());
        let capacity = std::num::NonZeroUsize::new(16).ok_or("capacity must be non-zero")?;
        let publisher = ActivityEventPublisher::new(
            store,
            capacity,
            crate::activity_publisher::TranscriptBatchPolicy::UNBATCHED,
        );
        let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();

        let pump = tokio::spawn(publish_declared_transcript(publisher.clone(), receiver));
        sender.send(line_event(0, "first"))?;
        sender.send(line_event(1, "second"))?;
        drop(sender);
        pump.await?;

        let key = ActivityStreamKey::of(&line_event(0, ""));
        let retained = publisher.replay_from(&key, 0).await?;
        let texts = retained
            .iter()
            .filter_map(|record| match &record.event.kind {
                ActivityEventKind::Message { text, .. } => Some(text.clone()),
                _ => None,
            })
            .collect::<Vec<_>>();
        assert_eq!(texts, vec!["first".to_owned(), "second".to_owned()]);
        Ok(())
    }
}