operonx 0.8.3

High-performance Rust execution backend for Operon workflows
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

//! Group / aggregate processors.
//!
//! Mirrors Python `processors/group.py`. Two related processors:
//!
//! - **`GroupBy`** — synthesises `GroupStart` / `GroupEnd` boundaries
//!   based on a `(op_name, payload-key)` predicate. Useful for callbot
//!   per-turn grouping where the turn boundary is implicit in the event
//!   stream (a `decider`-op annotation with `turn_id` changing).
//!
//! - **`Aggregate`** — collapses runs of same-kind, same-op events into a
//!   single summary event. Used to keep Langfuse observation count
//!   bounded when an op yields many frames.
//!
//! Both are stateful — they hold an in-flight group / aggregation across
//! a `process()` call. The Python impl is feature-complete; the Rust port
//! ships the public API + a no-op implementation for both, with a TODO to
//! flesh out the grouping logic when the callbot's Langfuse exporter port
//! exercises them. The default callbot pipeline doesn't compose these
//! processors so we're not gating Phase 2 on them.

use crate::core::tracing::events::TraceEvent;
use crate::core::tracing::pipeline::Processor;

/// Synthesise group boundary events based on `(op_name, payload-key)` shifts.
///
/// Phase-1 status: stub. The default callbot Langfuse exporter does its
/// own grouping; this processor isn't on the critical path. Port deferred
/// until a graph that needs it shows up.
pub struct GroupBy {
    pub op_name: String,
    pub key: String,
}

impl GroupBy {
    pub fn new(op_name: impl Into<String>, key: impl Into<String>) -> Self {
        Self {
            op_name: op_name.into(),
            key: key.into(),
        }
    }
}

impl Processor for GroupBy {
    fn name(&self) -> &'static str {
        "GroupBy"
    }
    fn process(&self, events: Vec<TraceEvent>) -> Vec<TraceEvent> {
        events
    }
}

/// Collapse runs of repeated same-kind same-op events.
///
/// Phase-1 status: stub (same rationale as `GroupBy`).
pub struct Aggregate {
    pub kinds: Vec<String>,
}

impl Aggregate {
    pub fn new(kinds: impl IntoIterator<Item = impl Into<String>>) -> Self {
        Self {
            kinds: kinds.into_iter().map(Into::into).collect(),
        }
    }
}

impl Processor for Aggregate {
    fn name(&self) -> &'static str {
        "Aggregate"
    }
    fn process(&self, events: Vec<TraceEvent>) -> Vec<TraceEvent> {
        events
    }
}