xagent-pi 0.2.10

Self-contained local brain (chat UI + API + SSE) for the Pi agent, tunneled into xagent-service.
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
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

//! Pi driver: spawn Pi, push events into the local brain, and feed the brain's
//! user prompts into Pi. (No XAGENT hub — purely local.)

use crate::brain::Brain;
use crate::pi::accumulator::Accumulator;
use crate::pi::converter::{self, Chunk};
use crate::pi::transport::PiTransport;
use anyhow::Result;
use serde_json::json;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;

pub struct PiDriver {
    transport: Arc<PiTransport>,
    brain: Brain,
    prompt_rx: mpsc::UnboundedReceiver<String>,
}

impl PiDriver {
    pub fn new(
        transport: PiTransport,
        brain: Brain,
        prompt_rx: mpsc::UnboundedReceiver<String>,
    ) -> Self {
        Self {
            transport: Arc::new(transport),
            brain,
            prompt_rx,
        }
    }

    pub async fn run(mut self) -> Result<()> {
        self.handshake().await;

        // Pi event pump → brain.
        let brain = self.brain.clone();
        let transport = Arc::clone(&self.transport);
        tokio::spawn(async move {
            pump(brain, transport).await;
        });

        // Brain prompts → Pi.
        while let Some(text) = self.prompt_rx.recv().await {
            log::info!("prompt -> pi ({} chars)", text.chars().count());
            if let Err(e) = self.transport.send(json!({ "type": "prompt", "message": text })) {
                log::warn!("pi send failed: {e}");
            }
        }
        Ok(())
    }

    async fn handshake(&self) {
        match self
            .transport
            .send_and_wait(json!({ "type": "get_state" }), Duration::from_secs(30))
            .await
        {
            Ok(resp) => {
                if let Some(model) = resp
                    .get("data")
                    .and_then(|d| d.get("model"))
                    .and_then(|m| m.get("id").or_else(|| m.get("modelId")))
                    .and_then(|s| s.as_str())
                {
                    log::info!("pi ready: model={model}");
                }
            }
            Err(e) => log::warn!("pi get_state failed ({e}); continuing"),
        }
    }
}

async fn pump(brain: Brain, transport: Arc<PiTransport>) {
    let mut acc = Accumulator::new();
    loop {
        let event = {
            let mut rx = transport.events_rx.lock().await;
            rx.recv().await
        };
        let Some(event) = event else {
            log::info!("pi events channel closed");
            break;
        };
        let t = event
            .get("type")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .to_string();

        for chunk in acc.handle(&event) {
            emit_chunk(&brain, chunk).await;
        }
        for chunk in converter::convert_event(&event) {
            emit_chunk(&brain, chunk).await;
        }

        match t.as_str() {
            "agent_start" | "turn_start" => brain.push_thinking(true).await,
            "turn_end" | "agent_end" => brain.push_thinking(false).await,
            _ => {}
        }
    }
}

async fn emit_chunk(brain: &Brain, chunk: Chunk) {
    match chunk {
        Chunk::Text(t) => brain.push_message("assistant", &t).await,
        Chunk::Reasoning(t) => brain.push_reasoning(&t).await,
        Chunk::ToolCall { id, name, input } => {
            brain.push_tool(&id, &name, Some(input), None, false).await
        }
        Chunk::ToolResult {
            id,
            name,
            result,
            is_error,
        } => brain
            .push_tool(&id, &name, None, Some(result), is_error)
            .await,
    }
}