adk-graph 2.3.0

//! Execution-log identity of a `#[task]` call.
//!
//! The execution log answers "has this call already run?" when a workflow
//! resumes and replays its entrypoint from the top. Keying it by the task's
//! name alone made every call of the same task the same call, so a task called
//! in a loop returned the first iteration's result for every later one.

use serde_json::Value;

/// Builds the execution-log key for one call of a task.
///
/// The key has the form `{task}#{ordinal}` or `{task}#{ordinal}:{arg_hash}`:
///
/// | Part | Meaning |
/// |------|---------|
/// | `task` | The task function's name. |
/// | `ordinal` | How many calls of this task preceded this one in the run, from `0`. A replay reaches the same calls in the same order, so the n-th call of a resumed run matches the n-th call of the run it resumes. |
/// | `arg_hash` | A stable 64-bit FNV-1a hash, as 16 hex digits, of the canonical JSON of the arguments. Present when at least one argument implements `Serialize`; an argument that does not is hashed as `null`. |
///
/// A replayed call whose arguments differ from the recorded call at the same
/// position gets a different key and runs again rather than returning another
/// call's result.
///
/// `#[task]` computes this through [`TaskContext`](super::TaskContext); call it
/// directly only to inspect an [`ExecutionLog`](super::ExecutionLog).
///
/// # Example
///
/// ```
/// use adk_graph::functional::task_call_id;
/// use serde_json::json;
///
/// assert_eq!(task_call_id("fetch", 0, &[]), "fetch#0");
///
/// let first = task_call_id("process", 0, &[Some(json!("a"))]);
/// let second = task_call_id("process", 1, &[Some(json!("b"))]);
/// assert_ne!(first, second);
/// assert!(first.starts_with("process#0:"));
/// ```
pub fn task_call_id(task_name: &str, ordinal: usize, args: &[Option<Value>]) -> String {
    if args.iter().all(Option::is_none) {
        return format!("{task_name}#{ordinal}");
    }
    let mut canonical = String::from("[");
    for (index, arg) in args.iter().enumerate() {
        if index > 0 {
            canonical.push(',');
        }
        write_canonical(arg.as_ref().unwrap_or(&Value::Null), &mut canonical);
    }
    canonical.push(']');
    let arg_hash = fnv1a_64(canonical.as_bytes());
    format!("{task_name}#{ordinal}:{arg_hash:016x}")
}

/// Writes JSON with object keys sorted, so a map argument hashes the same
/// whatever its iteration order.
fn write_canonical(value: &Value, out: &mut String) {
    match value {
        Value::Object(map) => {
            let mut keys: Vec<&String> = map.keys().collect();
            keys.sort();
            out.push('{');
            for (index, key) in keys.into_iter().enumerate() {
                if index > 0 {
                    out.push(',');
                }
                out.push_str(&Value::String(key.clone()).to_string());
                out.push(':');
                write_canonical(&map[key], out);
            }
            out.push('}');
        }
        Value::Array(items) => {
            out.push('[');
            for (index, item) in items.iter().enumerate() {
                if index > 0 {
                    out.push(',');
                }
                write_canonical(item, out);
            }
            out.push(']');
        }
        Value::Null | Value::Bool(_) | Value::Number(_) | Value::String(_) => {
            out.push_str(&value.to_string());
        }
    }
}

/// FNV-1a, chosen because its output is fixed by definition: the key must not
/// change between Rust releases, which rules out `std`'s `DefaultHasher`.
fn fnv1a_64(bytes: &[u8]) -> u64 {
    const OFFSET_BASIS: u64 = 0xcbf2_9ce4_8422_2325;
    const PRIME: u64 = 0x0000_0100_0000_01b3;
    bytes.iter().fold(OFFSET_BASIS, |hash, byte| (hash ^ u64::from(*byte)).wrapping_mul(PRIME))
}

/// Support for code generated by `#[task]`. Not a public API.
#[doc(hidden)]
pub mod __private {
    use serde::Serialize;
    use serde_json::Value;

    /// Wraps a task argument so method resolution picks [`SerializableTaskArg`]
    /// when the argument implements `Serialize` and [`OpaqueTaskArg`] otherwise.
    pub struct TaskArg<'a, T: ?Sized>(pub &'a T);

    /// Reads a serializable argument for the call's key.
    pub trait SerializableTaskArg {
        /// The argument as JSON, or `None` when it fails to serialize.
        fn task_arg_value(&self) -> Option<Value>;
    }

    impl<T: ?Sized + Serialize> SerializableTaskArg for TaskArg<'_, T> {
        fn task_arg_value(&self) -> Option<Value> {
            serde_json::to_value(self.0).ok()
        }
    }

    /// Fallback for an argument that does not implement `Serialize`.
    pub trait OpaqueTaskArg {
        /// Always `None`: the argument does not contribute to the key.
        fn task_arg_value(&self) -> Option<Value>;
    }

    impl<T: ?Sized> OpaqueTaskArg for &TaskArg<'_, T> {
        fn task_arg_value(&self) -> Option<Value> {
            None
        }
    }
}

#[cfg(test)]
mod tests {
    use super::__private::{OpaqueTaskArg, SerializableTaskArg, TaskArg};
    use super::*;
    use serde_json::json;

    #[test]
    fn a_call_without_serializable_arguments_is_keyed_by_ordinal() {
        assert_eq!(task_call_id("fetch", 0, &[]), "fetch#0");
        assert_eq!(task_call_id("fetch", 3, &[None, None]), "fetch#3");
    }

    #[test]
    fn the_key_is_stable_across_releases() {
        // Pinned: a change here orphans every execution log already persisted.
        assert_eq!(task_call_id("process", 1, &[Some(json!("a"))]), "process#1:2b288c997252f102");
    }

    #[test]
    fn different_arguments_give_different_keys() {
        let a = task_call_id("process", 0, &[Some(json!({"id": 1}))]);
        let b = task_call_id("process", 0, &[Some(json!({"id": 2}))]);
        assert_ne!(a, b);
    }

    #[test]
    fn map_key_order_does_not_change_the_key() {
        let mut forward = serde_json::Map::new();
        forward.insert("a".to_string(), json!(1));
        forward.insert("b".to_string(), json!([{"y": 2, "x": 1}]));
        let mut reverse = serde_json::Map::new();
        reverse.insert("b".to_string(), json!([{"x": 1, "y": 2}]));
        reverse.insert("a".to_string(), json!(1));

        assert_eq!(
            task_call_id("merge", 0, &[Some(Value::Object(forward))]),
            task_call_id("merge", 0, &[Some(Value::Object(reverse))]),
        );
    }

    // The explicit borrow is what selects between the two traits, exactly as in
    // the `#[task]` expansion; without it the opaque argument finds no method.
    #[allow(clippy::needless_borrow)]
    #[test]
    fn the_argument_probe_serializes_only_what_it_can() {
        struct NotSerializable;
        let serializable = "item";
        let opaque = NotSerializable;

        assert_eq!((&TaskArg(&serializable)).task_arg_value(), Some(json!("item")));
        assert_eq!((&TaskArg(&opaque)).task_arg_value(), None);
    }
}