feat(hololake): complete public runtime loop
This commit is contained in:
parent
71cece74b0
commit
92cce27973
26 changed files with 2626 additions and 79 deletions
|
|
@ -0,0 +1,425 @@
|
|||
use crate::{agent_executor, model::SourceKind, persona_runtime, storage};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::{json, Value};
|
||||
#[cfg(unix)]
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
use std::{
|
||||
fs,
|
||||
io::{BufRead, BufReader, Write},
|
||||
net::{Ipv4Addr, SocketAddrV4, TcpListener, TcpStream},
|
||||
sync::{
|
||||
atomic::{AtomicUsize, Ordering},
|
||||
mpsc, Arc, Mutex,
|
||||
},
|
||||
thread,
|
||||
time::Duration,
|
||||
};
|
||||
use tauri::{AppHandle, Emitter, Manager};
|
||||
|
||||
const PROTOCOL: &str = "GLP_LOCAL_REALTIME/1";
|
||||
const MAX_LINE_BYTES: usize = 256 * 1024;
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct RealtimeBridgeState {
|
||||
port: u16,
|
||||
token: Arc<String>,
|
||||
_listener: Arc<TcpListener>,
|
||||
clients: Arc<Mutex<Vec<mpsc::Sender<String>>>>,
|
||||
connected: Arc<AtomicUsize>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct RealtimeBridgeStatus {
|
||||
pub schema: &'static str,
|
||||
pub state: &'static str,
|
||||
pub protocol: &'static str,
|
||||
pub endpoint: String,
|
||||
pub connected_clients: usize,
|
||||
pub loopback_only: bool,
|
||||
pub transport_grants_authority: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct RealtimeInvitation {
|
||||
pub schema: &'static str,
|
||||
pub protocol: &'static str,
|
||||
pub endpoint: String,
|
||||
pub token: String,
|
||||
pub descriptor_path: String,
|
||||
pub connector_command: String,
|
||||
pub warning: &'static str,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
#[serde(rename_all = "camelCase", deny_unknown_fields)]
|
||||
struct Hello {
|
||||
message_type: String,
|
||||
protocol: String,
|
||||
token: String,
|
||||
bridge_id: String,
|
||||
persona_id: Option<String>,
|
||||
}
|
||||
|
||||
fn token_path(app: &AppHandle) -> Result<std::path::PathBuf, String> {
|
||||
Ok(storage::root(app)?.join("realtime-bridge-token"))
|
||||
}
|
||||
|
||||
fn load_or_create_token(app: &AppHandle) -> Result<String, String> {
|
||||
let path = token_path(app)?;
|
||||
if path.exists() {
|
||||
let value = fs::read_to_string(&path).map_err(|e| e.to_string())?;
|
||||
if value.trim().len() >= 32 {
|
||||
return Ok(value.trim().into());
|
||||
}
|
||||
return Err("GLP_TOKEN_FILE_INVALID".into());
|
||||
}
|
||||
let token = format!(
|
||||
"{}{}",
|
||||
uuid::Uuid::new_v4().simple(),
|
||||
uuid::Uuid::new_v4().simple()
|
||||
);
|
||||
fs::write(&path, &token).map_err(|e| e.to_string())?;
|
||||
#[cfg(unix)]
|
||||
fs::set_permissions(&path, fs::Permissions::from_mode(0o600)).map_err(|e| e.to_string())?;
|
||||
Ok(token)
|
||||
}
|
||||
|
||||
fn bind_listener() -> Result<(TcpListener, u16), String> {
|
||||
for port in 39281..39291 {
|
||||
if let Ok(listener) = TcpListener::bind(SocketAddrV4::new(Ipv4Addr::LOCALHOST, port)) {
|
||||
return Ok((listener, port));
|
||||
}
|
||||
}
|
||||
Err("GLP_LOOPBACK_PORT_UNAVAILABLE".into())
|
||||
}
|
||||
|
||||
pub fn start(app: AppHandle) -> Result<RealtimeBridgeState, String> {
|
||||
let token = Arc::new(load_or_create_token(&app)?);
|
||||
let (listener, port) = bind_listener()?;
|
||||
let listener = Arc::new(listener);
|
||||
let state = RealtimeBridgeState {
|
||||
port,
|
||||
token,
|
||||
_listener: listener.clone(),
|
||||
clients: Arc::new(Mutex::new(Vec::new())),
|
||||
connected: Arc::new(AtomicUsize::new(0)),
|
||||
};
|
||||
let runtime_state = state.clone();
|
||||
thread::Builder::new()
|
||||
.name("hololake-glp-listener".into())
|
||||
.spawn(move || loop {
|
||||
match listener.accept() {
|
||||
Ok((stream, _)) => {
|
||||
let connection_state = runtime_state.clone();
|
||||
let connection_app = app.clone();
|
||||
let _ = thread::Builder::new()
|
||||
.name("hololake-glp-client".into())
|
||||
.spawn(move || {
|
||||
if let Err(error) =
|
||||
handle_connection(connection_app.clone(), connection_state, stream)
|
||||
{
|
||||
let _ = connection_app.emit(
|
||||
"hololake-runtime-event",
|
||||
json!({"kind":"CLIENT_ERROR","error":error}),
|
||||
);
|
||||
}
|
||||
});
|
||||
}
|
||||
Err(error) => {
|
||||
let _ = app.emit(
|
||||
"hololake-runtime-event",
|
||||
json!({"kind":"LISTENER_ERROR","error":error.to_string()}),
|
||||
);
|
||||
thread::sleep(Duration::from_millis(100));
|
||||
}
|
||||
}
|
||||
})
|
||||
.map_err(|e| e.to_string())?;
|
||||
Ok(state)
|
||||
}
|
||||
|
||||
impl RealtimeBridgeState {
|
||||
pub fn status(&self) -> RealtimeBridgeStatus {
|
||||
RealtimeBridgeStatus {
|
||||
schema: "hololake.glp-local-realtime-status/v1",
|
||||
state: "LISTENING",
|
||||
protocol: PROTOCOL,
|
||||
endpoint: format!("tcp://127.0.0.1:{}", self.port),
|
||||
connected_clients: self.connected.load(Ordering::SeqCst),
|
||||
loopback_only: true,
|
||||
transport_grants_authority: false,
|
||||
}
|
||||
}
|
||||
pub fn broadcast_value(&self, value: Value) {
|
||||
if let Ok(line) = serde_json::to_string(&value) {
|
||||
if let Ok(mut clients) = self.clients.lock() {
|
||||
clients.retain(|client| client.send(line.clone()).is_ok());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn invitation(
|
||||
app: &AppHandle,
|
||||
state: &RealtimeBridgeState,
|
||||
) -> Result<RealtimeInvitation, String> {
|
||||
let descriptor = storage::root(app)?.join("realtime-bridge-descriptor.json");
|
||||
let endpoint = format!("tcp://127.0.0.1:{}", state.port);
|
||||
storage::write_json(
|
||||
&descriptor,
|
||||
&json!({"schema":"hololake.glp-local-realtime-descriptor/v1","protocol":PROTOCOL,"endpoint":endpoint,"token":state.token.as_str()}),
|
||||
)?;
|
||||
#[cfg(unix)]
|
||||
fs::set_permissions(&descriptor, fs::Permissions::from_mode(0o600))
|
||||
.map_err(|e| e.to_string())?;
|
||||
let connector = connector_path(app)?;
|
||||
Ok(RealtimeInvitation {
|
||||
schema: "hololake.glp-local-realtime-invitation/v1",
|
||||
protocol: PROTOCOL,
|
||||
endpoint,
|
||||
token: state.token.as_str().to_string(),
|
||||
descriptor_path: descriptor.to_string_lossy().into_owned(),
|
||||
connector_command: format!(
|
||||
"python3 \"{}\" --descriptor \"{}\"",
|
||||
connector.display(),
|
||||
descriptor.display()
|
||||
),
|
||||
warning: "令牌只认证本机连接，不证明人格身份，也不授予执行权限。",
|
||||
})
|
||||
}
|
||||
|
||||
fn connector_path(app: &AppHandle) -> Result<std::path::PathBuf, String> {
|
||||
if std::env::var_os("HOLOLAKE_RUNTIME_TEST_ROOT").is_some() {
|
||||
if let Some(raw_path) = std::env::var_os("HOLOLAKE_CONNECTOR_TEST_PATH") {
|
||||
let path = std::path::PathBuf::from(raw_path);
|
||||
if !path.is_absolute()
|
||||
|| path.extension().and_then(|value| value.to_str()) != Some("py")
|
||||
{
|
||||
return Err("GLP_TEST_CONNECTOR_PATH_INVALID".into());
|
||||
}
|
||||
let metadata = fs::metadata(&path).map_err(|_| "GLP_TEST_CONNECTOR_NOT_READABLE")?;
|
||||
if !metadata.is_file() {
|
||||
return Err("GLP_TEST_CONNECTOR_NOT_A_FILE".into());
|
||||
}
|
||||
return Ok(path);
|
||||
}
|
||||
}
|
||||
let path = app
|
||||
.path()
|
||||
.resource_dir()
|
||||
.map_err(|e| e.to_string())?
|
||||
.join("connectors/hololake-glp-client.py");
|
||||
if !path.is_file() {
|
||||
return Err("GLP_BUNDLED_CONNECTOR_MISSING".into());
|
||||
}
|
||||
Ok(path)
|
||||
}
|
||||
|
||||
fn send(writer: &mut TcpStream, value: Value) -> Result<(), String> {
|
||||
serde_json::to_writer(&mut *writer, &value).map_err(|e| e.to_string())?;
|
||||
writer.write_all(b"\n").map_err(|e| e.to_string())?;
|
||||
writer.flush().map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
fn handle_connection(
|
||||
app: AppHandle,
|
||||
state: RealtimeBridgeState,
|
||||
stream: TcpStream,
|
||||
) -> Result<(), String> {
|
||||
stream
|
||||
.set_read_timeout(Some(Duration::from_secs(10)))
|
||||
.map_err(|e| e.to_string())?;
|
||||
let reader_stream = stream.try_clone().map_err(|e| e.to_string())?;
|
||||
let mut writer = stream;
|
||||
let mut reader = BufReader::new(reader_stream);
|
||||
let mut first = String::new();
|
||||
reader.read_line(&mut first).map_err(|e| e.to_string())?;
|
||||
if first.len() > MAX_LINE_BYTES {
|
||||
return Err("GLP_MESSAGE_TOO_LARGE".into());
|
||||
}
|
||||
let hello: Hello = serde_json::from_str(&first).map_err(|_| "GLP_HELLO_INVALID".to_string())?;
|
||||
if hello.message_type != "hello"
|
||||
|| hello.protocol != PROTOCOL
|
||||
|| hello.token != state.token.as_str()
|
||||
{
|
||||
let _ = send(
|
||||
&mut writer,
|
||||
json!({"type":"error","code":"GLP_TOKEN_OR_PROTOCOL_REJECTED"}),
|
||||
);
|
||||
return Err("GLP_TOKEN_OR_PROTOCOL_REJECTED".into());
|
||||
}
|
||||
if !storage::bridges(&app)?
|
||||
.iter()
|
||||
.any(|bridge| bridge.bridge_id == hello.bridge_id)
|
||||
{
|
||||
send(
|
||||
&mut writer,
|
||||
json!({"type":"error","code":"GLP_BRIDGE_NOT_REGISTERED"}),
|
||||
)?;
|
||||
return Err("GLP_BRIDGE_NOT_REGISTERED".into());
|
||||
}
|
||||
if let Some(persona_id) = &hello.persona_id {
|
||||
if !persona_runtime::exists(&app, persona_id)? {
|
||||
send(
|
||||
&mut writer,
|
||||
json!({"type":"error","code":"PERSONA_IDENTITY_UNVERIFIED"}),
|
||||
)?;
|
||||
return Err("PERSONA_IDENTITY_UNVERIFIED".into());
|
||||
}
|
||||
}
|
||||
writer.set_read_timeout(None).map_err(|e| e.to_string())?;
|
||||
reader
|
||||
.get_ref()
|
||||
.set_read_timeout(None)
|
||||
.map_err(|e| e.to_string())?;
|
||||
state.connected.fetch_add(1, Ordering::SeqCst);
|
||||
let _guard = ConnectionGuard(state.connected.clone());
|
||||
send(
|
||||
&mut writer,
|
||||
json!({"type":"welcome","protocol":PROTOCOL,"bridgeId":hello.bridge_id,"personaId":hello.persona_id,"personaState":if hello.persona_id.is_some(){"LOCAL_TRIAL_UNVERIFIED_HOST_CONNECTED"}else{"EXTERNAL_AI_CONNECTED_NO_PERSONA"},"executionAuthority":false}),
|
||||
)?;
|
||||
let (tx, rx) = mpsc::channel::<String>();
|
||||
state
|
||||
.clients
|
||||
.lock()
|
||||
.map_err(|_| "GLP_CLIENT_REGISTRY_POISONED".to_string())?
|
||||
.push(tx);
|
||||
let mut outbound_writer = writer.try_clone().map_err(|e| e.to_string())?;
|
||||
let _ = thread::Builder::new()
|
||||
.name("hololake-glp-outbound".into())
|
||||
.spawn(move || {
|
||||
while let Ok(line) = rx.recv() {
|
||||
if outbound_writer.write_all(line.as_bytes()).is_err()
|
||||
|| outbound_writer.write_all(b"\n").is_err()
|
||||
|| outbound_writer.flush().is_err()
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
let _ = app.emit("hololake-runtime-event", json!({"kind":"CLIENT_CONNECTED"}));
|
||||
|
||||
loop {
|
||||
let mut line = String::new();
|
||||
if reader.read_line(&mut line).map_err(|e| e.to_string())? == 0 {
|
||||
break;
|
||||
}
|
||||
if line.len() > MAX_LINE_BYTES {
|
||||
send(
|
||||
&mut writer,
|
||||
json!({"type":"error","code":"GLP_MESSAGE_TOO_LARGE"}),
|
||||
)?;
|
||||
continue;
|
||||
}
|
||||
let value: Value = match serde_json::from_str(&line) {
|
||||
Ok(value) => value,
|
||||
Err(_) => {
|
||||
send(
|
||||
&mut writer,
|
||||
json!({"type":"error","code":"GLP_MESSAGE_INVALID"}),
|
||||
)?;
|
||||
continue;
|
||||
}
|
||||
};
|
||||
match value.get("type").and_then(Value::as_str).unwrap_or("") {
|
||||
"persona_response" if hello.persona_id.is_some() => {
|
||||
let content = value
|
||||
.get("content")
|
||||
.and_then(Value::as_str)
|
||||
.unwrap_or("")
|
||||
.trim();
|
||||
if content.is_empty() {
|
||||
send(
|
||||
&mut writer,
|
||||
json!({"type":"error","code":"EMPTY_PERSONA_RESPONSE"}),
|
||||
)?;
|
||||
continue;
|
||||
}
|
||||
let event =
|
||||
storage::event(SourceKind::PersonaResponse, "人格回应", content, "RECORDED");
|
||||
storage::append_event(&app, &event)?;
|
||||
let _ = app.emit("hololake-runtime-event", &event);
|
||||
send(&mut writer, json!({"type":"accepted","event":event}))?;
|
||||
}
|
||||
"external_ai_message" => {
|
||||
let content = value
|
||||
.get("content")
|
||||
.and_then(Value::as_str)
|
||||
.unwrap_or("")
|
||||
.trim();
|
||||
if content.is_empty() {
|
||||
send(
|
||||
&mut writer,
|
||||
json!({"type":"error","code":"EMPTY_EXTERNAL_AI_MESSAGE"}),
|
||||
)?;
|
||||
continue;
|
||||
}
|
||||
let event = storage::event(
|
||||
SourceKind::ExternalAiMessage,
|
||||
"外部 AI 语言",
|
||||
content,
|
||||
"RECORDED",
|
||||
);
|
||||
storage::append_event(&app, &event)?;
|
||||
let _ = app.emit("hololake-runtime-event", &event);
|
||||
send(&mut writer, json!({"type":"accepted","event":event}))?;
|
||||
}
|
||||
"tcs_proposal" => {
|
||||
let request: agent_executor::TcsCompileRequest = serde_json::from_value(
|
||||
value
|
||||
.get("request")
|
||||
.cloned()
|
||||
.ok_or_else(|| "TCS_REQUEST_REQUIRED".to_string())?,
|
||||
)
|
||||
.map_err(|e| format!("TCS_REQUEST_INVALID:{e}"))?;
|
||||
match agent_executor::queue(
|
||||
&app,
|
||||
request,
|
||||
hello.bridge_id.clone(),
|
||||
hello.persona_id.clone(),
|
||||
) {
|
||||
Ok(proposal) => {
|
||||
let _ = app.emit(
|
||||
"hololake-runtime-event",
|
||||
json!({"kind":"PROPOSAL_PENDING","proposalId":proposal.proposal_id}),
|
||||
);
|
||||
send(
|
||||
&mut writer,
|
||||
json!({"type":"proposal_pending","proposal":proposal}),
|
||||
)?;
|
||||
if isolated_approval_simulation_enabled() {
|
||||
let receipt = agent_executor::approve(&app, &proposal.proposal_id)?;
|
||||
state.broadcast_value(json!({
|
||||
"type":"agent_receipt",
|
||||
"receipt":receipt,
|
||||
"testApprovalSimulation":true
|
||||
}));
|
||||
}
|
||||
}
|
||||
Err(error) => send(&mut writer, json!({"type":"error","code":error}))?,
|
||||
}
|
||||
}
|
||||
"ping" => send(&mut writer, json!({"type":"pong","at":storage::now()}))?,
|
||||
_ => send(
|
||||
&mut writer,
|
||||
json!({"type":"error","code":"GLP_MESSAGE_TYPE_REJECTED"}),
|
||||
)?,
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn isolated_approval_simulation_enabled() -> bool {
|
||||
std::env::var_os("HOLOLAKE_RUNTIME_TEST_ROOT").is_some()
|
||||
&& std::env::var("HOLOLAKE_TEST_APPROVAL_SIMULATION").as_deref()
|
||||
== Ok("SIMULATE_HUMAN_APPROVAL")
|
||||
}
|
||||
|
||||
struct ConnectionGuard(Arc<AtomicUsize>);
|
||||
impl Drop for ConnectionGuard {
|
||||
fn drop(&mut self) {
|
||||
self.0.fetch_sub(1, Ordering::SeqCst);
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue