gproxy_protocol/adapt/generate/fanout/
mod.rs

1//! Bounded multi-candidate generation over concrete single-result edges.
2//! Progress lives only in memory: a started child without an actual result is
3//! never replayed, because nothing reloads a group once its invocation ends.
4
5mod edges;
6mod invoke;
7pub(super) mod output;
8mod prepare;
9use super::{Endpoint, GenerationIdentity, GenerationProgress};
10use crate::{
11    codec::CodecLimits,
12    transform::{
13        TransformError, TransformErrorKind,
14        identity::{IdNamespace, IdentityFlow, IdentityRole, SourceIdentity, TargetIdPolicy},
15    },
16};
17pub use prepare::{
18    ChatViaClaudeFanout, ChatViaResponsesFanout, FanoutTarget, GeminiViaClaudeFanout,
19    GeminiViaResponsesFanout,
20};
21
22/// Host-supplied invocation namespace. Child namespaces are derived from it.
23#[derive(Debug, Clone)]
24pub struct FanoutOptions {
25    pub namespace: IdNamespace,
26
27    pub response_policy: TargetIdPolicy,
28}
29
30/// Caller-owned evidence survives cancellation and failed post-send state writes,
31/// so a caller can inspect which children were sent and what they returned. It
32/// is never persisted and cannot restart a group: a prepared fanout sends at
33/// most once, so a started POST is never sent twice.
34#[derive(Debug)]
35pub struct FanoutProgress<N> {
36    children: Vec<GenerationProgress<N>>,
37}
38
39impl<N> Default for FanoutProgress<N> {
40    fn default() -> Self {
41        Self { children: vec![] }
42    }
43}
44
45impl<N> FanoutProgress<N> {
46    pub fn children(&self) -> &[GenerationProgress<N>] {
47        &self.children
48    }
49}
50
51#[derive(Debug)]
52struct Fanout<A> {
53    children: Vec<A>,
54    endpoint: Endpoint,
55    group_id: String,
56    /// Set once the first child may have been sent. Nothing persists a group,
57    /// so this flag is what keeps a second `invoke` from repeating its POSTs.
58    started: bool,
59}
60
61fn conflict(message: impl Into<String>) -> TransformError {
62    TransformError::new(TransformErrorKind::Conflict, "generation.fanout", message)
63}
64
65fn limit() -> TransformError {
66    TransformError::new(
67        TransformErrorKind::Limit,
68        "generation.fanout",
69        "fanout budget exceeded",
70    )
71}
72
73fn codec_error(e: crate::codec::CodecError) -> TransformError {
74    TransformError::new(
75        if e.kind() == crate::codec::CodecErrorKind::Limit {
76            TransformErrorKind::Limit
77        } else {
78            TransformErrorKind::InvalidResult
79        },
80        "generation.fanout",
81        e.to_string(),
82    )
83}
84
85pub(super) fn group_id(options: &FanoutOptions) -> Result<String, TransformError> {
86    IdentityFlow::new(options.namespace)
87        .resolve_or_allocate(
88            IdentityRole::Response,
89            SourceIdentity::new(options.response_policy.dialect, None, 0),
90            &options.response_policy,
91        )
92        .map(|v| v.emitted_id)
93        .map_err(|e| conflict(e.to_string()))
94}
95
96impl FanoutOptions {
97    pub(in crate::adapt::generate) fn child_identity(
98        &self,
99        index: i64,
100        upstream: crate::Dialect,
101    ) -> GenerationIdentity {
102        let namespace = |lane: u8| {
103            let mut hash = blake3::Hasher::new_derive_key("gproxy.fanout.child.v1");
104            hash.update(&self.namespace.bytes());
105            hash.update(&index.to_le_bytes());
106            hash.update(&[lane]);
107            let mut bytes = [0; 16];
108            bytes.copy_from_slice(&hash.finalize().as_bytes()[..16]);
109            IdNamespace(bytes)
110        };
111        GenerationIdentity {
112            request: IdentityFlow::new(namespace(0)),
113            response: IdentityFlow::new(namespace(1)),
114            request_policy: super::transport::generation_policy(upstream),
115            response_policy: self.response_policy.clone().with_reversible_tool_calls(),
116        }
117    }
118}
119
120fn encode<T: serde::Serialize>(v: &T, limits: CodecLimits) -> Result<Vec<u8>, TransformError> {
121    crate::codec::encode_json(v, limits)
122        .map(|b| b.to_vec())
123        .map_err(codec_error)
124}
gproxy_protocol/adapt/generate/fanout/mod.rs

gproxy_protocol/adapt/generate/fanout/
mod.rs
