bingshuo/hololake-system-architecture
Watch
1
0
Fork
You've already forked hololake-system-architecture
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity

feat(hololake): add resumable direct local AI broker

Browse source
This commit is contained in:
冰朔 2026-08-13 12:21:04 +08:00
commit 9d1801266c
10 changed files with 1289 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

445
product-source/hololake-native-desktop/src-tauri/src/direct_local_broker.rs Normal file
View file

@ -0,0 +1,445 @@
use crate::direct_local_session::{
append_event_at, direct_session_root, open_at, resume_at, AppendSessionEventInput,
OpenSessionInput, ResumeSessionInput,
};
use ring::digest::{digest, SHA256};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::fs::{self, OpenOptions};
use std::io::{self, BufRead, BufReader, Read, Write};
use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::sync::{
atomic::{AtomicBool, Ordering},
Arc,
};
use std::thread;
use std::time::Duration;
use tauri::{AppHandle, Manager};
use uuid::Uuid;
const BROKER_SCHEMA: &str = "hololake.direct-local-broker/v1";
const MAX_REQUEST_BYTES: u64 = 1024 * 1024;
pub struct DirectLocalBrokerHandle {
shutdown: Arc<AtomicBool>,
socket_path: PathBuf,
}
impl Drop for DirectLocalBrokerHandle {
fn drop(&mut self) {
self.shutdown.store(true, Ordering::Release);
let _ = UnixStream::connect(&self.socket_path);
}
}
#[derive(Debug, Deserialize)]
#[serde(
tag = "operation",
content = "input",
rename_all = "SCREAMING_SNAKE_CASE"
)]
enum BrokerRequest {
Ping,
OpenSession(OpenSessionInput),
ResumeSession(ResumeSessionInput),
AppendEvent(AppendSessionEventInput),
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct BrokerResponse {
schema: &'static str,
request_id: String,
ok: bool,
result: Option<Value>,
error: Option<String>,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct BrokerDescriptor {
schema: &'static str,
state: &'static str,
transport: &'static str,
socket_path: String,
max_request_bytes: u64,
process_id: u32,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct ConnectorDescriptor {
schema: String,
state: String,
transport: String,
socket_path: String,
}
pub fn start(app: &AppHandle) -> Result<DirectLocalBrokerHandle, Box<dyn std::error::Error>> {
let session_root = direct_session_root(app).map_err(std::io::Error::other)?;
let app_data = app.path().app_data_dir()?;
let descriptor_path = app_data.join("direct-local-broker-v1.json");
let socket_path = short_socket_path(&app_data);
start_at(session_root, descriptor_path, socket_path).map_err(|error| error.into())
}
pub fn run_connector() -> Result<(), String> {
let descriptor_path = connector_descriptor_path()?;
let descriptor: ConnectorDescriptor = serde_json::from_slice(
&fs::read(&descriptor_path)
.map_err(|error| format!("HOLOLAKE_BROKER_DESCRIPTOR_UNAVAILABLE: {error}"))?,
)
.map_err(|error| format!("HOLOLAKE_BROKER_DESCRIPTOR_INVALID: {error}"))?;
if descriptor.schema != BROKER_SCHEMA
|| descriptor.state != "LISTENING"
|| descriptor.transport != "UNIX_STREAM_JSON_LINES"
{
return Err("HOLOLAKE_BROKER_DESCRIPTOR_UNSUPPORTED".into());
}
let mut stream = UnixStream::connect(&descriptor.socket_path)
.map_err(|error| format!("HOLOLAKE_DIRECT_LOCAL_BROKER_NOT_RUNNING: {error}"))?;
let read_stream = stream
.try_clone()
.map_err(|error| format!("HOLOLAKE_CONNECTOR_CLONE_FAILED: {error}"))?;
let output = thread::Builder::new()
.name("hololake-connector-output".into())
.spawn(move || -> Result<(), String> {
let mut reader = BufReader::new(read_stream);
let mut stdout = io::stdout().lock();
io::copy(&mut reader, &mut stdout)
.and_then(|_| stdout.flush())
.map_err(|error| format!("HOLOLAKE_CONNECTOR_OUTPUT_FAILED: {error}"))?;
Ok(())
})
.map_err(|error| format!("HOLOLAKE_CONNECTOR_OUTPUT_THREAD_FAILED: {error}"))?;
let mut stdin = io::stdin().lock();
io::copy(&mut stdin, &mut stream)
.and_then(|_| stream.shutdown(std::net::Shutdown::Write))
.map_err(|error| format!("HOLOLAKE_CONNECTOR_INPUT_FAILED: {error}"))?;
output
.join()
.map_err(|_| "HOLOLAKE_CONNECTOR_OUTPUT_THREAD_PANICKED".to_string())??;
Ok(())
}
fn connector_descriptor_path() -> Result<PathBuf, String> {
if let Some(path) = std::env::var_os("HOLOLAKE_BROKER_DESCRIPTOR") {
return Ok(PathBuf::from(path));
}
dirs::data_dir()
.map(|root| {
root.join("world.guanghu.hololake")
.join("direct-local-broker-v1.json")
})
.ok_or_else(|| "HOLOLAKE_APP_DATA_UNAVAILABLE".into())
}
fn start_at(
session_root: PathBuf,
descriptor_path: PathBuf,
socket_path: PathBuf,
) -> Result<DirectLocalBrokerHandle, String> {
if let Some(parent) = socket_path.parent() {
fs::create_dir_all(parent)
.map_err(|error| format!("HOLOLAKE_BROKER_RUNTIME_DIR_FAILED: {error}"))?;
fs::set_permissions(parent, fs::Permissions::from_mode(0o700))
.map_err(|error| format!("HOLOLAKE_BROKER_RUNTIME_PERMISSIONS_FAILED: {error}"))?;
}
if socket_path.exists() {
if UnixStream::connect(&socket_path).is_ok() {
return Err("HOLOLAKE_DIRECT_LOCAL_BROKER_ALREADY_RUNNING".into());
}
fs::remove_file(&socket_path)
.map_err(|error| format!("HOLOLAKE_BROKER_STALE_SOCKET_REMOVE_FAILED: {error}"))?;
}
let listener = UnixListener::bind(&socket_path)
.map_err(|error| format!("HOLOLAKE_BROKER_BIND_FAILED: {error}"))?;
fs::set_permissions(&socket_path, fs::Permissions::from_mode(0o600))
.map_err(|error| format!("HOLOLAKE_BROKER_SOCKET_PERMISSIONS_FAILED: {error}"))?;
listener
.set_nonblocking(true)
.map_err(|error| format!("HOLOLAKE_BROKER_NONBLOCKING_FAILED: {error}"))?;
write_descriptor(
&descriptor_path,
&BrokerDescriptor {
schema: BROKER_SCHEMA,
state: "LISTENING",
transport: "UNIX_STREAM_JSON_LINES",
socket_path: socket_path.to_string_lossy().into_owned(),
max_request_bytes: MAX_REQUEST_BYTES,
process_id: std::process::id(),
},
)?;
let shutdown = Arc::new(AtomicBool::new(false));
let worker_shutdown = Arc::clone(&shutdown);
let worker_socket = socket_path.clone();
thread::Builder::new()
.name("hololake-direct-local-broker".into())
.spawn(move || {
serve(listener, session_root, &worker_shutdown);
let _ = fs::remove_file(worker_socket);
})
.map_err(|error| format!("HOLOLAKE_BROKER_THREAD_FAILED: {error}"))?;
Ok(DirectLocalBrokerHandle {
shutdown,
socket_path,
})
}
fn serve(listener: UnixListener, session_root: PathBuf, shutdown: &AtomicBool) {
while !shutdown.load(Ordering::Acquire) {
match listener.accept() {
Ok((stream, _)) => {
let root = session_root.clone();
let _ = thread::Builder::new()
.name("hololake-direct-local-client".into())
.spawn(move || serve_connection(stream, &root));
}
Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
thread::sleep(Duration::from_millis(25));
}
Err(_) => break,
}
}
}
fn serve_connection(mut stream: UnixStream, session_root: &Path) {
let read_stream = match stream.try_clone() {
Ok(stream) => stream,
Err(_) => return,
};
let mut reader = BufReader::new(read_stream);
loop {
let mut bytes = Vec::new();
let read = match reader
.by_ref()
.take(MAX_REQUEST_BYTES + 1)
.read_until(b'\n', &mut bytes)
{
Ok(read) => read,
Err(_) => return,
};
if read == 0 {
return;
}
let response = if read as u64 > MAX_REQUEST_BYTES || !bytes.ends_with(b"\n") {
BrokerResponse::error("HOLOLAKE_BROKER_REQUEST_TOO_LARGE")
} else {
dispatch(session_root, &bytes[..bytes.len() - 1])
};
if serde_json::to_writer(&mut stream, &response).is_err()
|| stream.write_all(b"\n").is_err()
|| stream.flush().is_err()
{
return;
}
if response.error.as_deref() == Some("HOLOLAKE_BROKER_REQUEST_TOO_LARGE") {
return;
}
}
}
fn dispatch(session_root: &Path, bytes: &[u8]) -> BrokerResponse {
let request: BrokerRequest = match serde_json::from_slice(bytes) {
Ok(request) => request,
Err(error) => {
return BrokerResponse::error(&format!("HOLOLAKE_BROKER_REQUEST_INVALID: {error}"))
}
};
let result = match request {
BrokerRequest::Ping => serde_json::to_value(serde_json::json!({
"state": "READY",
"continuityOwner": "HOLOLAKE",
"mcpRole": "DISCOVERY_RECOVERY_COMPATIBILITY_ONLY"
}))
.map_err(|error| error.to_string()),
BrokerRequest::OpenSession(input) => open_at(session_root, input)
.and_then(|receipt| serde_json::to_value(receipt).map_err(|error| error.to_string())),
BrokerRequest::ResumeSession(input) => resume_at(session_root, input)
.and_then(|receipt| serde_json::to_value(receipt).map_err(|error| error.to_string())),
BrokerRequest::AppendEvent(input) => append_event_at(session_root, input)
.and_then(|receipt| serde_json::to_value(receipt).map_err(|error| error.to_string())),
};
match result {
Ok(value) => BrokerResponse::success(value),
Err(error) => BrokerResponse::error(&error),
}
}
impl BrokerResponse {
fn success(result: Value) -> Self {
Self {
schema: BROKER_SCHEMA,
request_id: Uuid::new_v4().to_string(),
ok: true,
result: Some(result),
error: None,
}
}
fn error(error: &str) -> Self {
Self {
schema: BROKER_SCHEMA,
request_id: Uuid::new_v4().to_string(),
ok: false,
result: None,
error: Some(error.into()),
}
}
}
fn short_socket_path(app_data: &Path) -> PathBuf {
let key = digest(&SHA256, app_data.to_string_lossy().as_bytes())
.as_ref()
.iter()
.take(8)
.map(|byte| format!("{byte:02x}"))
.collect::<String>();
std::env::temp_dir()
.join(format!("hololake-{key}"))
.join("broker.sock")
}
fn write_descriptor(path: &Path, descriptor: &BrokerDescriptor) -> Result<(), String> {
let parent = path
.parent()
.ok_or("HOLOLAKE_BROKER_DESCRIPTOR_PATH_INVALID")?;
fs::create_dir_all(parent)
.map_err(|error| format!("HOLOLAKE_BROKER_DESCRIPTOR_DIR_FAILED: {error}"))?;
let temporary = parent.join(format!(".broker-{}.tmp", Uuid::new_v4()));
let bytes = serde_json::to_vec_pretty(descriptor)
.map_err(|error| format!("HOLOLAKE_BROKER_DESCRIPTOR_INVALID: {error}"))?;
let mut file = OpenOptions::new()
.write(true)
.create_new(true)
.mode(0o600)
.open(&temporary)
.map_err(|error| format!("HOLOLAKE_BROKER_DESCRIPTOR_WRITE_FAILED: {error}"))?;
file.write_all(&bytes)
.and_then(|_| file.sync_all())
.map_err(|error| format!("HOLOLAKE_BROKER_DESCRIPTOR_WRITE_FAILED: {error}"))?;
fs::rename(&temporary, path)
.map_err(|error| format!("HOLOLAKE_BROKER_DESCRIPTOR_WRITE_FAILED: {error}"))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::direct_local_session::{issue_ticket_at, IssueDiscoveryTicketInput};
use tempfile::TempDir;
fn request(socket: &Path, value: Value) -> Value {
let mut stream = UnixStream::connect(socket).unwrap();
serde_json::to_writer(&mut stream, &value).unwrap();
stream.write_all(b"\n").unwrap();
let mut line = String::new();
BufReader::new(stream).read_line(&mut line).unwrap();
serde_json::from_str(&line).unwrap()
}
#[test]
fn broker_resumes_after_socket_disconnect() {
let temp = TempDir::new().unwrap();
let socket = temp.path().join("runtime/broker.sock");
let descriptor = temp.path().join("broker.json");
let sessions = temp.path().join("sessions");
fs::create_dir_all(&sessions).unwrap();
let _broker = start_at(sessions.clone(), descriptor, socket.clone()).unwrap();
let ticket = issue_ticket_at(
&sessions,
IssueDiscoveryTicketInput {
account_id: "human-1".into(),
lane_id: "DEV-1".into(),
client_instance_id: "codex-1".into(),
},
)
.unwrap();
let opened = request(
&socket,
serde_json::json!({
"operation": "OPEN_SESSION",
"input": {
"accountId": "human-1",
"laneId": "DEV-1",
"clientInstanceId": "codex-1",
"discoveryTicket": ticket.discovery_ticket
}
}),
);
assert_eq!(opened["ok"], true);
let resumed = request(
&socket,
serde_json::json!({
"operation": "RESUME_SESSION",
"input": {
"accountId": "human-1",
"sessionId": opened["result"]["sessionId"],
"clientInstanceId": "codex-restarted",
"resumeSecret": opened["result"]["resumeSecret"]
}
}),
);
assert_eq!(resumed["ok"], true);
assert_eq!(resumed["result"]["state"], "RESUMED");
}
#[test]
fn broker_can_restart_and_resume_persisted_session() {
let temp = TempDir::new().unwrap();
let sessions = temp.path().join("sessions");
fs::create_dir_all(&sessions).unwrap();
let socket = temp.path().join("runtime/broker.sock");
let descriptor = temp.path().join("broker.json");
let ticket = issue_ticket_at(
&sessions,
IssueDiscoveryTicketInput {
account_id: "human-1".into(),
lane_id: "DEV-1".into(),
client_instance_id: "codex-1".into(),
},
)
.unwrap();
let first = start_at(sessions.clone(), descriptor.clone(), socket.clone()).unwrap();
let opened = request(
&socket,
serde_json::json!({
"operation": "OPEN_SESSION",
"input": {
"accountId": "human-1",
"laneId": "DEV-1",
"clientInstanceId": "codex-1",
"discoveryTicket": ticket.discovery_ticket
}
}),
);
drop(first);
for _ in 0..40 {
if !socket.exists() {
break;
}
thread::sleep(Duration::from_millis(10));
}
let _second = start_at(sessions, descriptor, socket.clone()).unwrap();
let resumed = request(
&socket,
serde_json::json!({
"operation": "RESUME_SESSION",
"input": {
"accountId": "human-1",
"sessionId": opened["result"]["sessionId"],
"clientInstanceId": "codex-after-hololake-restart",
"resumeSecret": opened["result"]["resumeSecret"]
}
}),
);
assert_eq!(resumed["result"]["state"], "RESUMED");
}
}

790
product-source/hololake-native-desktop/src-tauri/src/direct_local_session.rs Normal file
View file

@ -0,0 +1,790 @@
use fs2::FileExt;
use ring::digest::{digest, SHA256};
use serde::{Deserialize, Serialize};
use std::fs::{self, OpenOptions};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::time::{SystemTime, UNIX_EPOCH};
use tauri::{AppHandle, Manager};
use uuid::Uuid;
const SESSION_SCHEMA: &str = "hololake.direct-local-session/v1";
const MAX_ID_BYTES: usize = 128;
#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct OpenSessionInput {
pub account_id: String,
pub lane_id: String,
pub client_instance_id: String,
pub discovery_ticket: String,
}
#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct IssueDiscoveryTicketInput {
pub account_id: String,
pub lane_id: String,
pub client_instance_id: String,
}
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct DiscoveryTicketReceipt {
pub schema: &'static str,
pub state: &'static str,
pub account_key: String,
pub lane_id: String,
pub client_instance_id: String,
pub discovery_ticket: String,
pub issued_at_unix_ms: u128,
pub receipt_id: String,
}
#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ResumeSessionInput {
pub account_id: String,
pub session_id: String,
pub client_instance_id: String,
pub resume_secret: String,
}
#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AppendSessionEventInput {
pub account_id: String,
pub session_id: String,
pub resume_secret: String,
pub idempotency_key: String,
pub event_kind: String,
pub payload_sha256: String,
pub expected_previous_sequence: u64,
}
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct DirectSessionReceipt {
pub schema: &'static str,
pub state: &'static str,
pub account_key: String,
pub session_id: String,
pub lane_id: String,
pub client_instance_id: String,
pub opened_at_unix_ms: u128,
pub observed_at_unix_ms: u128,
pub last_event_sequence: u64,
pub resume_secret: Option<String>,
pub receipt_id: String,
}
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct SessionEventReceipt {
pub schema: &'static str,
pub state: &'static str,
pub session_id: String,
pub sequence: u64,
pub idempotency_key: String,
pub event_kind: String,
pub payload_sha256: String,
pub observed_at_unix_ms: u128,
pub receipt_id: String,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct SessionRecord {
schema: String,
account_key: String,
session_id: String,
lane_id: String,
client_instance_id: String,
resume_secret_sha256: String,
discovery_ticket_sha256: String,
opened_at_unix_ms: u128,
observed_at_unix_ms: u128,
last_event_sequence: u64,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct EventRecord {
schema: String,
session_id: String,
sequence: u64,
idempotency_key: String,
event_kind: String,
payload_sha256: String,
observed_at_unix_ms: u128,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct DiscoveryTicketRecord {
schema: String,
account_key: String,
lane_id: String,
client_instance_id: String,
discovery_ticket_sha256: String,
issued_at_unix_ms: u128,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct ActiveSessionRecord {
schema: String,
account_key: String,
session_id: String,
lane_id: String,
}
#[tauri::command]
pub async fn issue_direct_local_discovery_ticket(
app: AppHandle,
input: IssueDiscoveryTicketInput,
) -> Result<DiscoveryTicketReceipt, String> {
let root = direct_session_root(&app)?;
tauri::async_runtime::spawn_blocking(move || issue_ticket_at(&root, input))
.await
.map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_JOIN_FAILED: {error}"))?
}
#[tauri::command]
pub async fn open_direct_local_session(
app: AppHandle,
input: OpenSessionInput,
) -> Result<DirectSessionReceipt, String> {
let root = direct_session_root(&app)?;
tauri::async_runtime::spawn_blocking(move || open_at(&root, input))
.await
.map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_JOIN_FAILED: {error}"))?
}
#[tauri::command]
pub async fn resume_direct_local_session(
app: AppHandle,
input: ResumeSessionInput,
) -> Result<DirectSessionReceipt, String> {
let root = direct_session_root(&app)?;
tauri::async_runtime::spawn_blocking(move || resume_at(&root, input))
.await
.map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_JOIN_FAILED: {error}"))?
}
#[tauri::command]
pub async fn append_direct_local_session_event(
app: AppHandle,
input: AppendSessionEventInput,
) -> Result<SessionEventReceipt, String> {
let root = direct_session_root(&app)?;
tauri::async_runtime::spawn_blocking(move || append_event_at(&root, input))
.await
.map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_JOIN_FAILED: {error}"))?
}
pub(crate) fn direct_session_root(app: &AppHandle) -> Result<PathBuf, String> {
let app_data = app
.path()
.app_data_dir()
.map_err(|error| format!("HOLOLAKE_APP_DATA_UNAVAILABLE: {error}"))?;
let root = app_data.join("direct-local-session-v1");
fs::create_dir_all(&root)
.map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?;
root.canonicalize()
.map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))
}
pub(crate) fn open_at(
root: &Path,
input: OpenSessionInput,
) -> Result<DirectSessionReceipt, String> {
validate_identifier(&input.account_id, "ACCOUNT")?;
validate_identifier(&input.lane_id, "LANE")?;
validate_identifier(&input.client_instance_id, "CLIENT")?;
validate_secret(&input.discovery_ticket, "DISCOVERY_TICKET")?;
let account_key = sha256_hex(input.account_id.as_bytes());
let _account_lock = lock_account(root, &account_key)?;
let active_path = active_session_path(root, &account_key);
let tickets_root = root.join("discovery-tickets");
let used_tickets_root = root.join("used-discovery-tickets");
fs::create_dir_all(&used_tickets_root)
.map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?;
let discovery_ticket_sha256 = sha256_hex(input.discovery_ticket.as_bytes());
let ticket_path = tickets_root.join(format!("{discovery_ticket_sha256}.json"));
let used_ticket_path = used_tickets_root.join(format!("{discovery_ticket_sha256}.json"));
if used_ticket_path.exists() {
return Err("HOLOLAKE_DISCOVERY_TICKET_ALREADY_USED".into());
}
if active_path.exists() {
let active: ActiveSessionRecord = read_json(&active_path, "ACTIVE_SESSION")?;
if active.schema != SESSION_SCHEMA || active.account_key != account_key {
return Err("HOLOLAKE_ACTIVE_SESSION_INVALID".into());
}
return Err(format!(
"HOLOLAKE_ACCOUNT_ALREADY_HAS_ACTIVE_SESSION:{}:{}",
active.lane_id, active.session_id
));
}
let ticket: DiscoveryTicketRecord = read_json(&ticket_path, "DISCOVERY_TICKET")?;
if ticket.account_key != account_key
|| ticket.lane_id != input.lane_id
|| ticket.client_instance_id != input.client_instance_id
|| ticket.discovery_ticket_sha256 != discovery_ticket_sha256
{
return Err("HOLOLAKE_DISCOVERY_TICKET_NOT_AUTHORIZED".into());
}
fs::rename(&ticket_path, &used_ticket_path).map_err(|error| {
if error.kind() == std::io::ErrorKind::NotFound && used_ticket_path.exists() {
"HOLOLAKE_DISCOVERY_TICKET_ALREADY_USED".to_string()
} else {
format!("HOLOLAKE_DISCOVERY_TICKET_CONSUME_FAILED: {error}")
}
})?;
let session_id = Uuid::new_v4().to_string();
let resume_secret = format!("{}{}", Uuid::new_v4(), Uuid::new_v4());
let observed_at_unix_ms = now_unix_ms()?;
let record = SessionRecord {
schema: SESSION_SCHEMA.into(),
account_key: account_key.clone(),
session_id: session_id.clone(),
lane_id: input.lane_id,
client_instance_id: input.client_instance_id,
resume_secret_sha256: sha256_hex(resume_secret.as_bytes()),
discovery_ticket_sha256,
opened_at_unix_ms: observed_at_unix_ms,
observed_at_unix_ms,
last_event_sequence: 0,
};
write_record_atomic(&session_path(root, &account_key, &session_id), &record)?;
write_record_create_new(
&active_path,
&ActiveSessionRecord {
schema: SESSION_SCHEMA.into(),
account_key: account_key.clone(),
session_id: session_id.clone(),
lane_id: record.lane_id.clone(),
},
"ACTIVE_SESSION",
)?;
Ok(session_receipt("OPENED", &record, Some(resume_secret)))
}
pub(crate) fn issue_ticket_at(
root: &Path,
input: IssueDiscoveryTicketInput,
) -> Result<DiscoveryTicketReceipt, String> {
validate_identifier(&input.account_id, "ACCOUNT")?;
validate_identifier(&input.lane_id, "LANE")?;
validate_identifier(&input.client_instance_id, "CLIENT")?;
let account_key = sha256_hex(input.account_id.as_bytes());
let discovery_ticket = format!("{}{}", Uuid::new_v4(), Uuid::new_v4());
let discovery_ticket_sha256 = sha256_hex(discovery_ticket.as_bytes());
let issued_at_unix_ms = now_unix_ms()?;
let record = DiscoveryTicketRecord {
schema: SESSION_SCHEMA.into(),
account_key: account_key.clone(),
lane_id: input.lane_id.clone(),
client_instance_id: input.client_instance_id.clone(),
discovery_ticket_sha256: discovery_ticket_sha256.clone(),
issued_at_unix_ms,
};
let path = root
.join("discovery-tickets")
.join(format!("{discovery_ticket_sha256}.json"));
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)
.map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?;
}
write_record_create_new(&path, &record, "DISCOVERY_TICKET")?;
Ok(DiscoveryTicketReceipt {
schema: SESSION_SCHEMA,
state: "ISSUED",
account_key,
lane_id: input.lane_id,
client_instance_id: input.client_instance_id,
discovery_ticket,
issued_at_unix_ms,
receipt_id: sha256_hex(
format!("ISSUED\n{discovery_ticket_sha256}\n{issued_at_unix_ms}").as_bytes(),
),
})
}
pub(crate) fn resume_at(
root: &Path,
input: ResumeSessionInput,
) -> Result<DirectSessionReceipt, String> {
validate_identifier(&input.account_id, "ACCOUNT")?;
validate_identifier(&input.session_id, "SESSION")?;
validate_identifier(&input.client_instance_id, "CLIENT")?;
validate_secret(&input.resume_secret, "RESUME_SECRET")?;
let account_key = sha256_hex(input.account_id.as_bytes());
require_active_session(root, &account_key, &input.session_id)?;
let path = session_path(root, &account_key, &input.session_id);
let _lock = lock_session(&path)?;
let mut record = read_session(&path)?;
authorize(
&record,
&account_key,
&input.session_id,
&input.resume_secret,
)?;
record.client_instance_id = input.client_instance_id;
record.observed_at_unix_ms = now_unix_ms()?;
write_record_atomic(&path, &record)?;
Ok(session_receipt("RESUMED", &record, None))
}
pub(crate) fn append_event_at(
root: &Path,
input: AppendSessionEventInput,
) -> Result<SessionEventReceipt, String> {
validate_identifier(&input.account_id, "ACCOUNT")?;
validate_identifier(&input.session_id, "SESSION")?;
validate_identifier(&input.idempotency_key, "IDEMPOTENCY")?;
validate_identifier(&input.event_kind, "EVENT_KIND")?;
validate_sha256(&input.payload_sha256)?;
validate_secret(&input.resume_secret, "RESUME_SECRET")?;
let account_key = sha256_hex(input.account_id.as_bytes());
require_active_session(root, &account_key, &input.session_id)?;
let path = session_path(root, &account_key, &input.session_id);
let _lock = lock_session(&path)?;
let mut record = read_session(&path)?;
authorize(
&record,
&account_key,
&input.session_id,
&input.resume_secret,
)?;
let events_root = path
.parent()
.ok_or("HOLOLAKE_DIRECT_SESSION_PATH_INVALID")?
.join("events");
fs::create_dir_all(&events_root)
.map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?;
let idempotency_path = events_root.join(format!("idempotency-{}.json", input.idempotency_key));
if idempotency_path.exists() {
let existing: EventRecord = read_json(&idempotency_path, "EVENT")?;
if existing.event_kind == input.event_kind
&& existing.payload_sha256 == input.payload_sha256
{
return Ok(event_receipt("DUPLICATE_CONFIRMED", &existing));
}
return Err("HOLOLAKE_EVENT_IDEMPOTENCY_CONFLICT".into());
}
if input.expected_previous_sequence != record.last_event_sequence {
return Err("HOLOLAKE_EVENT_CURSOR_CONFLICT".into());
}
let observed_at_unix_ms = now_unix_ms()?;
let event = EventRecord {
schema: SESSION_SCHEMA.into(),
session_id: record.session_id.clone(),
sequence: record.last_event_sequence + 1,
idempotency_key: input.idempotency_key,
event_kind: input.event_kind,
payload_sha256: input.payload_sha256,
observed_at_unix_ms,
};
write_record_create_new(&idempotency_path, &event, "EVENT")?;
record.last_event_sequence = event.sequence;
record.observed_at_unix_ms = observed_at_unix_ms;
write_record_atomic(&path, &record)?;
Ok(event_receipt("APPENDED", &event))
}
fn session_path(root: &Path, account_key: &str, session_id: &str) -> PathBuf {
root.join("accounts")
.join(account_key)
.join("sessions")
.join(session_id)
.join("session.json")
}
fn active_session_path(root: &Path, account_key: &str) -> PathBuf {
root.join("accounts")
.join(account_key)
.join("active-session.json")
}
fn require_active_session(root: &Path, account_key: &str, session_id: &str) -> Result<(), String> {
let active: ActiveSessionRecord =
read_json(&active_session_path(root, account_key), "ACTIVE_SESSION")?;
if active.schema != SESSION_SCHEMA
|| active.account_key != account_key
|| active.session_id != session_id
{
return Err("HOLOLAKE_SESSION_IS_NOT_ACCOUNT_ACTIVE_SESSION".into());
}
Ok(())
}
fn lock_account(root: &Path, account_key: &str) -> Result<std::fs::File, String> {
lock_file(
&root.join("accounts").join(account_key).join("account.lock"),
"ACCOUNT",
)
}
fn read_session(path: &Path) -> Result<SessionRecord, String> {
let record: SessionRecord = read_json(path, "SESSION")?;
if record.schema != SESSION_SCHEMA {
return Err("HOLOLAKE_DIRECT_SESSION_SCHEMA_UNSUPPORTED".into());
}
Ok(record)
}
fn lock_session(path: &Path) -> Result<std::fs::File, String> {
let parent = path
.parent()
.ok_or("HOLOLAKE_DIRECT_SESSION_PATH_INVALID")?;
fs::create_dir_all(parent)
.map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?;
lock_file(&parent.join("session.lock"), "SESSION")
}
fn lock_file(lock_path: &Path, kind: &str) -> Result<std::fs::File, String> {
let parent = lock_path
.parent()
.ok_or("HOLOLAKE_DIRECT_SESSION_PATH_INVALID")?;
fs::create_dir_all(parent)
.map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?;
let file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(lock_path)
.map_err(|error| format!("HOLOLAKE_DIRECT_{kind}_LOCK_FAILED: {error}"))?;
file.lock_exclusive()
.map_err(|error| format!("HOLOLAKE_DIRECT_{kind}_LOCK_FAILED: {error}"))?;
Ok(file)
}
fn authorize(
record: &SessionRecord,
account_key: &str,
session_id: &str,
secret: &str,
) -> Result<(), String> {
if record.account_key != account_key
|| record.session_id != session_id
|| record.resume_secret_sha256 != sha256_hex(secret.as_bytes())
{
return Err("HOLOLAKE_DIRECT_SESSION_NOT_AUTHORIZED".into());
}
Ok(())
}
fn write_record_atomic<T: Serialize>(path: &Path, record: &T) -> Result<(), String> {
let parent = path
.parent()
.ok_or("HOLOLAKE_DIRECT_SESSION_PATH_INVALID")?;
fs::create_dir_all(parent)
.map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?;
let temporary = parent.join(format!(".{}.tmp", Uuid::new_v4()));
write_bytes_create_new(&temporary, record, "SESSION")?;
fs::rename(&temporary, path)
.map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_WRITE_FAILED: {error}"))
}
fn write_record_create_new<T: Serialize>(
path: &Path,
record: &T,
kind: &str,
) -> Result<(), String> {
write_bytes_create_new(path, record, kind)
}
fn write_bytes_create_new<T: Serialize>(path: &Path, record: &T, kind: &str) -> Result<(), String> {
let bytes = serde_json::to_vec_pretty(record)
.map_err(|error| format!("HOLOLAKE_DIRECT_{kind}_INVALID: {error}"))?;
let mut file = OpenOptions::new()
.write(true)
.create_new(true)
.open(path)
.map_err(|error| format!("HOLOLAKE_DIRECT_{kind}_WRITE_FAILED: {error}"))?;
file.write_all(&bytes)
.and_then(|_| file.sync_all())
.map_err(|error| format!("HOLOLAKE_DIRECT_{kind}_WRITE_FAILED: {error}"))
}
fn read_json<T: for<'de> Deserialize<'de>>(path: &Path, kind: &str) -> Result<T, String> {
let bytes = fs::read(path).map_err(|error| {
if error.kind() == std::io::ErrorKind::NotFound {
format!("HOLOLAKE_DIRECT_{kind}_NOT_FOUND")
} else {
format!("HOLOLAKE_DIRECT_{kind}_UNREADABLE: {error}")
}
})?;
serde_json::from_slice(&bytes)
.map_err(|error| format!("HOLOLAKE_DIRECT_{kind}_INVALID: {error}"))
}
fn session_receipt(
state: &'static str,
record: &SessionRecord,
resume_secret: Option<String>,
) -> DirectSessionReceipt {
DirectSessionReceipt {
schema: SESSION_SCHEMA,
state,
account_key: record.account_key.clone(),
session_id: record.session_id.clone(),
lane_id: record.lane_id.clone(),
client_instance_id: record.client_instance_id.clone(),
opened_at_unix_ms: record.opened_at_unix_ms,
observed_at_unix_ms: record.observed_at_unix_ms,
last_event_sequence: record.last_event_sequence,
resume_secret,
receipt_id: sha256_hex(
format!(
"{state}\n{}\n{}\n{}\n{}",
record.account_key, record.session_id, record.lane_id, record.observed_at_unix_ms
)
.as_bytes(),
),
}
}
fn event_receipt(state: &'static str, event: &EventRecord) -> SessionEventReceipt {
SessionEventReceipt {
schema: SESSION_SCHEMA,
state,
session_id: event.session_id.clone(),
sequence: event.sequence,
idempotency_key: event.idempotency_key.clone(),
event_kind: event.event_kind.clone(),
payload_sha256: event.payload_sha256.clone(),
observed_at_unix_ms: event.observed_at_unix_ms,
receipt_id: sha256_hex(
format!(
"{}\n{}\n{}\n{}",
event.session_id, event.sequence, event.idempotency_key, event.payload_sha256
)
.as_bytes(),
),
}
}
fn validate_identifier(value: &str, kind: &str) -> Result<(), String> {
if value.is_empty()
|| value.len() > MAX_ID_BYTES
|| !value
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b':'))
{
return Err(format!("HOLOLAKE_{kind}_ID_INVALID"));
}
Ok(())
}
fn validate_secret(value: &str, kind: &str) -> Result<(), String> {
if value.len() < 16 || value.len() > 512 {
return Err(format!("HOLOLAKE_{kind}_INVALID"));
}
Ok(())
}
fn validate_sha256(value: &str) -> Result<(), String> {
if value.len() != 64
|| !value
.bytes()
.all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
{
return Err("HOLOLAKE_PAYLOAD_SHA256_INVALID".into());
}
Ok(())
}
fn now_unix_ms() -> Result<u128, String> {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_millis())
.map_err(|error| format!("HOLOLAKE_SYSTEM_CLOCK_INVALID: {error}"))
}
fn sha256_hex(value: &[u8]) -> String {
digest(&SHA256, value)
.as_ref()
.iter()
.map(|byte| format!("{byte:02x}"))
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::TempDir;
fn open_input(ticket: &str) -> OpenSessionInput {
OpenSessionInput {
account_id: "human-BS-0001".into(),
lane_id: "DEV-001".into(),
client_instance_id: "codex-1".into(),
discovery_ticket: ticket.into(),
}
}
fn issued_open_input(temp: &TempDir, suffix: &str) -> OpenSessionInput {
let issued = issue_ticket_at(
temp.path(),
IssueDiscoveryTicketInput {
account_id: "human-BS-0001".into(),
lane_id: "DEV-001".into(),
client_instance_id: "codex-1".into(),
},
)
.unwrap();
let mut input = open_input(&issued.discovery_ticket);
input.client_instance_id = format!("codex-{suffix}");
if suffix != "1" {
let issued = issue_ticket_at(
temp.path(),
IssueDiscoveryTicketInput {
account_id: input.account_id.clone(),
lane_id: input.lane_id.clone(),
client_instance_id: input.client_instance_id.clone(),
},
)
.unwrap();
input.discovery_ticket = issued.discovery_ticket;
}
input
}
#[test]
fn discovery_ticket_is_single_use_and_secret_is_not_stored_plaintext() {
let temp = TempDir::new().unwrap();
let input = issued_open_input(&temp, "1");
let ticket = input.discovery_ticket.clone();
let opened = open_at(temp.path(), input).unwrap();
assert_eq!(opened.state, "OPENED");
assert!(open_at(temp.path(), open_input(&ticket))
.unwrap_err()
.contains("ALREADY_USED"));
let stored = fs::read_to_string(session_path(
temp.path(),
&opened.account_key,
&opened.session_id,
))
.unwrap();
assert!(!stored.contains(opened.resume_secret.as_ref().unwrap()));
}
#[test]
fn ai_restart_resumes_the_same_session_and_cursor() {
let temp = TempDir::new().unwrap();
let opened = open_at(temp.path(), issued_open_input(&temp, "2")).unwrap();
let event = append_event_at(
temp.path(),
AppendSessionEventInput {
account_id: "human-BS-0001".into(),
session_id: opened.session_id.clone(),
resume_secret: opened.resume_secret.clone().unwrap(),
idempotency_key: "event-1".into(),
event_kind: "TASK_OBSERVED".into(),
payload_sha256: "a".repeat(64),
expected_previous_sequence: 0,
},
)
.unwrap();
assert_eq!(event.sequence, 1);
let resumed = resume_at(
temp.path(),
ResumeSessionInput {
account_id: "human-BS-0001".into(),
session_id: opened.session_id,
client_instance_id: "codex-2".into(),
resume_secret: opened.resume_secret.unwrap(),
},
)
.unwrap();
assert_eq!(resumed.state, "RESUMED");
assert_eq!(resumed.last_event_sequence, 1);
assert_eq!(resumed.client_instance_id, "codex-2");
}
#[test]
fn event_retries_are_idempotent_but_conflicts_fail_closed() {
let temp = TempDir::new().unwrap();
let opened = open_at(temp.path(), issued_open_input(&temp, "3")).unwrap();
let secret = opened.resume_secret.unwrap();
let input = AppendSessionEventInput {
account_id: "human-BS-0001".into(),
session_id: opened.session_id,
resume_secret: secret,
idempotency_key: "event-1".into(),
event_kind: "TASK_OBSERVED".into(),
payload_sha256: "b".repeat(64),
expected_previous_sequence: 0,
};
assert_eq!(
append_event_at(temp.path(), input.clone()).unwrap().state,
"APPENDED"
);
assert_eq!(
append_event_at(temp.path(), input.clone()).unwrap().state,
"DUPLICATE_CONFIRMED"
);
let conflicting = AppendSessionEventInput {
payload_sha256: "c".repeat(64),
..input
};
assert_eq!(
append_event_at(temp.path(), conflicting).unwrap_err(),
"HOLOLAKE_EVENT_IDEMPOTENCY_CONFLICT"
);
}
#[test]
fn wrong_secret_and_stale_cursor_fail_closed() {
let temp = TempDir::new().unwrap();
let opened = open_at(temp.path(), issued_open_input(&temp, "4")).unwrap();
let wrong = resume_at(
temp.path(),
ResumeSessionInput {
account_id: "human-BS-0001".into(),
session_id: opened.session_id.clone(),
client_instance_id: "codex-2".into(),
resume_secret: "wrong-resume-secret-0000".into(),
},
);
assert_eq!(wrong.unwrap_err(), "HOLOLAKE_DIRECT_SESSION_NOT_AUTHORIZED");
let stale = append_event_at(
temp.path(),
AppendSessionEventInput {
account_id: "human-BS-0001".into(),
session_id: opened.session_id,
resume_secret: opened.resume_secret.unwrap(),
idempotency_key: "event-2".into(),
event_kind: "TASK_OBSERVED".into(),
payload_sha256: "d".repeat(64),
expected_previous_sequence: 9,
},
);
assert_eq!(stale.unwrap_err(), "HOLOLAKE_EVENT_CURSOR_CONFLICT");
}
#[test]
fn caller_cannot_invent_a_discovery_ticket() {
let temp = TempDir::new().unwrap();
assert_eq!(
open_at(temp.path(), open_input("invented-ticket-0000")).unwrap_err(),
"HOLOLAKE_DIRECT_DISCOVERY_TICKET_NOT_FOUND"
);
}
#[test]
fn one_account_cannot_open_two_direct_write_sessions() {
let temp = TempDir::new().unwrap();
let first = open_at(temp.path(), issued_open_input(&temp, "1")).unwrap();
let second = open_at(temp.path(), issued_open_input(&temp, "2")).unwrap_err();
assert!(second.contains("ACCOUNT_ALREADY_HAS_ACTIVE_SESSION"));
assert!(second.contains(&first.session_id));
}
}

14
product-source/hololake-native-desktop/src-tauri/src/lib.rs
View file

@ -1,15 +1,29 @@
mod direct_local_broker;
mod direct_local_session;
mod local_development_bridge;
mod release_trust;
use tauri::Manager;
pub fn run_connector() -> Result<(), String> {
direct_local_broker::run_connector()
}
#[cfg_attr(mobile, tauri::mobile_entry_point)]
pub fn run() {
tauri::Builder::default()
.invoke_handler(tauri::generate_handler![
direct_local_session::issue_direct_local_discovery_ticket,
direct_local_session::open_direct_local_session,
direct_local_session::resume_direct_local_session,
direct_local_session::append_direct_local_session_event,
local_development_bridge::acquire_development_write_lane,
local_development_bridge::inspect_development_write_lane,
local_development_bridge::release_development_write_lane,
])
.setup(|app| {
let broker = direct_local_broker::start(app.handle())?;
app.manage(broker);
release_trust::install_updater_if_provisioned(app.handle())?;
Ok(())
})

7
product-source/hololake-native-desktop/src-tauri/src/main.rs
View file

@ -1,5 +1,12 @@
#![cfg_attr(not(debug_assertions), windows_subsystem = "windows")]
fn main() {
if std::env::args().any(|argument| argument == "--connector") {
if let Err(error) = hololake_native_desktop_lib::run_connector() {
eprintln!("{error}");
std::process::exit(1);
}
return;
}
hololake_native_desktop_lib::run();
}