feat: add incremental persona remote object channel
Human-Responsibility: ICE-GL∞ / 冰朔 Persona-Author: ICE-P-ZY001 / 铸渊 Execution-Runtime: Codex desktop / DEV-20260810-014 Development-ID: DEV-20260810-014 Authorization-Scope: GH-PNCC persona runtime source and tests only; no UI, deployment, or execution limb Source-Anchor: daily continuity appends to the existing life chain without redownloading all history
This commit is contained in:
铸渊 / ICE-P-ZY001
parent
30d179e97a
commit
1b55f3407a
13 changed files with 1111 additions and 4 deletions
|
|
@ -37,6 +37,7 @@ mod opencode_config;
|
|||
mod opencode_discovery;
|
||||
mod opencode_events;
|
||||
mod persona_code_channel;
|
||||
mod persona_remote_git;
|
||||
pub mod pi_cli;
|
||||
mod pi_config;
|
||||
mod pi_discovery;
|
||||
|
|
@ -533,6 +534,7 @@ macro_rules! app_invoke_handler {
|
|||
persona_code_channel::recover_persona_code_channel_lifecycle_request,
|
||||
persona_code_channel::inspect_persona_code_channel_session,
|
||||
persona_code_channel::recover_persona_code_channel_session,
|
||||
persona_remote_git::read_persona_remote_git_object,
|
||||
guanghu_router::guanghu_router_connect,
|
||||
guanghu_router::guanghu_router_disconnect,
|
||||
guanghu_router::guanghu_router_approve,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,871 @@
|
|||
use ring::digest::{digest, SHA256};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::fs::{self, OpenOptions};
|
||||
use std::io::Write;
|
||||
use std::path::{Component, Path, PathBuf};
|
||||
use std::process::{Command, Output};
|
||||
use uuid::Uuid;
|
||||
|
||||
const DEFAULT_BRANCH: &str = "main";
|
||||
const DEFAULT_CACHE_LIMIT_BYTES: u64 = 128 * 1024 * 1024;
|
||||
const MIN_CACHE_LIMIT_BYTES: u64 = 1024 * 1024;
|
||||
const MAX_REMOTE_TEXT_BYTES: usize = 2 * 1024 * 1024;
|
||||
const MAX_CONTINUITY_REHYDRATION_DEPTH: usize = 4096;
|
||||
|
||||
#[derive(Clone, Debug, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct PersonaRemoteReadInput {
|
||||
pub remote_url: String,
|
||||
pub branch: Option<String>,
|
||||
pub relative_path: String,
|
||||
pub expected_head: Option<String>,
|
||||
pub cache_root: String,
|
||||
pub max_cache_bytes: Option<u64>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct PersonaRemoteReadReceipt {
|
||||
pub schema: &'static str,
|
||||
pub remote_head: String,
|
||||
pub previous_cursor: Option<String>,
|
||||
pub cursor_advanced: bool,
|
||||
pub continuity_verified: bool,
|
||||
pub continuity_state_path: String,
|
||||
pub fetched_incremental_objects: bool,
|
||||
pub cache_rehydrated: bool,
|
||||
pub relative_path: String,
|
||||
pub content: String,
|
||||
pub content_sha256: String,
|
||||
pub cache_path: String,
|
||||
pub cache_bytes: u64,
|
||||
pub cache_limit_bytes: u64,
|
||||
pub cache_state: &'static str,
|
||||
pub worktree_created: bool,
|
||||
pub full_history_requested: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct PersonaRemoteCursorRecord {
|
||||
schema: String,
|
||||
remote_identity_hash: String,
|
||||
branch: String,
|
||||
verified_head: String,
|
||||
previous_verified_head: Option<String>,
|
||||
generation: u64,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
struct RemoteReadRequest {
|
||||
remote_url: String,
|
||||
branch: String,
|
||||
relative_path: String,
|
||||
expected_head: Option<String>,
|
||||
cache_root: PathBuf,
|
||||
max_cache_bytes: u64,
|
||||
allow_file_remote: bool,
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn read_persona_remote_git_object(
|
||||
input: PersonaRemoteReadInput,
|
||||
) -> Result<PersonaRemoteReadReceipt, String> {
|
||||
let request = RemoteReadRequest::try_from(input)?;
|
||||
tokio::task::spawn_blocking(move || read_remote_text_object(&request))
|
||||
.await
|
||||
.map_err(|error| format!("PERSONA_REMOTE_READ_JOIN_FAILED: {error}"))?
|
||||
}
|
||||
|
||||
impl TryFrom<PersonaRemoteReadInput> for RemoteReadRequest {
|
||||
type Error = String;
|
||||
|
||||
fn try_from(input: PersonaRemoteReadInput) -> Result<Self, Self::Error> {
|
||||
let branch = input.branch.unwrap_or_else(|| DEFAULT_BRANCH.to_string());
|
||||
validate_remote_url(&input.remote_url, false)?;
|
||||
validate_branch(&branch)?;
|
||||
validate_relative_path(&input.relative_path)?;
|
||||
let expected_head = input
|
||||
.expected_head
|
||||
.map(|head| validate_head(&head).map(str::to_owned))
|
||||
.transpose()?;
|
||||
let max_cache_bytes = input.max_cache_bytes.unwrap_or(DEFAULT_CACHE_LIMIT_BYTES);
|
||||
if max_cache_bytes < MIN_CACHE_LIMIT_BYTES {
|
||||
return Err("PERSONA_REMOTE_CACHE_LIMIT_TOO_SMALL".into());
|
||||
}
|
||||
let cache_root = prepare_cache_root(Path::new(&input.cache_root))?;
|
||||
Ok(Self {
|
||||
remote_url: input.remote_url,
|
||||
branch,
|
||||
relative_path: input.relative_path,
|
||||
expected_head,
|
||||
cache_root,
|
||||
max_cache_bytes,
|
||||
allow_file_remote: false,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
fn read_remote_text_object(
|
||||
request: &RemoteReadRequest,
|
||||
) -> Result<PersonaRemoteReadReceipt, String> {
|
||||
validate_remote_url(&request.remote_url, request.allow_file_remote)?;
|
||||
validate_branch(&request.branch)?;
|
||||
validate_relative_path(&request.relative_path)?;
|
||||
|
||||
let remote_head = read_remote_head(&request.remote_url, &request.branch)?;
|
||||
if request
|
||||
.expected_head
|
||||
.as_deref()
|
||||
.is_some_and(|expected| expected != remote_head)
|
||||
{
|
||||
return Err("PERSONA_REMOTE_HEAD_MISMATCH".into());
|
||||
}
|
||||
|
||||
let identity_hash = cache_key(&request.remote_url, &request.branch);
|
||||
let object_cache_root = prepare_child_root(&request.cache_root, "objects")?;
|
||||
let continuity_root = prepare_child_root(&request.cache_root, "continuity")?;
|
||||
let cache_path = object_cache_root.join(format!("{identity_hash}.git"));
|
||||
let continuity_state_path = continuity_root.join(format!("{identity_hash}.json"));
|
||||
let cursor_record =
|
||||
read_cursor_record(&continuity_state_path, &identity_hash, &request.branch)?;
|
||||
let previous_cursor = cursor_record
|
||||
.as_ref()
|
||||
.map(|record| record.verified_head.clone());
|
||||
|
||||
ensure_bare_partial_cache(&cache_path, &request.remote_url)?;
|
||||
let remote_ref = format!("refs/remotes/origin/{}", request.branch);
|
||||
let cached_cursor = optional_ref(&cache_path, &remote_ref)?;
|
||||
match (previous_cursor.as_deref(), cached_cursor.as_deref()) {
|
||||
(None, Some(_)) => return Err("PERSONA_REMOTE_CACHE_WITHOUT_CONTINUITY_STATE".into()),
|
||||
(Some(expected), Some(observed)) if expected != observed => {
|
||||
return Err("PERSONA_REMOTE_CACHE_CURSOR_MISMATCH".into())
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
|
||||
let cursor_advanced = previous_cursor.as_deref() != Some(remote_head.as_str());
|
||||
let cache_rehydrated = cached_cursor.is_none() && previous_cursor.is_some();
|
||||
let fetched_incremental_objects = cached_cursor.as_deref() != Some(remote_head.as_str());
|
||||
if fetched_incremental_objects {
|
||||
fetch_remote_cursor(
|
||||
&cache_path,
|
||||
&request.branch,
|
||||
previous_cursor.as_deref(),
|
||||
cached_cursor.is_some(),
|
||||
)?;
|
||||
}
|
||||
let fetched_head = required_ref(&cache_path, &remote_ref)?;
|
||||
if fetched_head != remote_head {
|
||||
return Err("PERSONA_REMOTE_FETCH_CURSOR_MISMATCH".into());
|
||||
}
|
||||
if let Some(previous) = previous_cursor.as_deref() {
|
||||
if previous != remote_head && !git_is_ancestor(&cache_path, previous, &remote_head)? {
|
||||
remove_exact_cache(&object_cache_root, &cache_path)?;
|
||||
return Err("PERSONA_REMOTE_HISTORY_REWRITE_REJECTED".into());
|
||||
}
|
||||
}
|
||||
|
||||
let object = format!("{remote_head}:{}", request.relative_path);
|
||||
let content_bytes = git_output_bytes(
|
||||
git_at(&cache_path).args(["show", &object]),
|
||||
"PERSONA_REMOTE_OBJECT_READ",
|
||||
)?;
|
||||
if content_bytes.len() > MAX_REMOTE_TEXT_BYTES {
|
||||
return Err("PERSONA_REMOTE_OBJECT_TOO_LARGE".into());
|
||||
}
|
||||
let content = String::from_utf8(content_bytes)
|
||||
.map_err(|_| "PERSONA_REMOTE_OBJECT_NOT_UTF8".to_string())?;
|
||||
let content_sha256 = sha256_hex(content.as_bytes());
|
||||
|
||||
let next_record = PersonaRemoteCursorRecord {
|
||||
schema: "hololake.persona-remote-git-cursor/v1".into(),
|
||||
remote_identity_hash: identity_hash,
|
||||
branch: request.branch.clone(),
|
||||
verified_head: remote_head.clone(),
|
||||
previous_verified_head: previous_cursor.clone(),
|
||||
generation: cursor_record
|
||||
.as_ref()
|
||||
.map(|record| record.generation.saturating_add(u64::from(cursor_advanced)))
|
||||
.unwrap_or(1),
|
||||
};
|
||||
write_cursor_record(&continuity_state_path, &next_record)?;
|
||||
|
||||
let observed_cache_bytes = directory_size(&cache_path)?;
|
||||
let (cache_bytes, cache_state) = if observed_cache_bytes > request.max_cache_bytes {
|
||||
remove_exact_cache(&object_cache_root, &cache_path)?;
|
||||
(0, "EVICTED_AFTER_BOUNDED_READ")
|
||||
} else {
|
||||
(observed_cache_bytes, "BOUNDED_PARTIAL_OBJECT_CACHE")
|
||||
};
|
||||
|
||||
Ok(PersonaRemoteReadReceipt {
|
||||
schema: "hololake.persona-remote-git-object-read/v1",
|
||||
remote_head,
|
||||
previous_cursor,
|
||||
cursor_advanced,
|
||||
continuity_verified: true,
|
||||
continuity_state_path: continuity_state_path.to_string_lossy().into_owned(),
|
||||
fetched_incremental_objects,
|
||||
cache_rehydrated,
|
||||
relative_path: request.relative_path.clone(),
|
||||
content,
|
||||
content_sha256,
|
||||
cache_path: cache_path.to_string_lossy().into_owned(),
|
||||
cache_bytes,
|
||||
cache_limit_bytes: request.max_cache_bytes,
|
||||
cache_state,
|
||||
worktree_created: false,
|
||||
full_history_requested: false,
|
||||
})
|
||||
}
|
||||
|
||||
fn validate_remote_url(remote_url: &str, allow_file_remote: bool) -> Result<(), String> {
|
||||
let is_file_remote = allow_file_remote && remote_url.starts_with("file://");
|
||||
let accepted_scheme = remote_url.starts_with("https://") || is_file_remote;
|
||||
if !accepted_scheme || remote_url.contains(['\n', '\r', '\0', '?', '#']) {
|
||||
return Err("PERSONA_REMOTE_URL_UNSUPPORTED".into());
|
||||
}
|
||||
let authority = remote_url
|
||||
.split_once("://")
|
||||
.map(|(_, rest)| rest.split('/').next().unwrap_or_default())
|
||||
.unwrap_or_default();
|
||||
if (!is_file_remote && authority.is_empty()) || authority.contains('@') {
|
||||
return Err("PERSONA_REMOTE_URL_MUST_NOT_CONTAIN_CREDENTIALS".into());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn validate_branch(branch: &str) -> Result<(), String> {
|
||||
if branch.is_empty() || branch.len() > 200 || branch.starts_with('-') {
|
||||
return Err("PERSONA_REMOTE_BRANCH_INVALID".into());
|
||||
}
|
||||
let output = Command::new("git")
|
||||
.args(["check-ref-format", "--branch", branch])
|
||||
.output()
|
||||
.map_err(|error| format!("PERSONA_REMOTE_GIT_UNAVAILABLE: {error}"))?;
|
||||
if output.status.success() {
|
||||
Ok(())
|
||||
} else {
|
||||
Err("PERSONA_REMOTE_BRANCH_INVALID".into())
|
||||
}
|
||||
}
|
||||
|
||||
fn validate_relative_path(relative: &str) -> Result<(), String> {
|
||||
let path = Path::new(relative);
|
||||
if relative.is_empty()
|
||||
|| relative.len() > 1024
|
||||
|| path.is_absolute()
|
||||
|| path.components().any(|component| {
|
||||
matches!(
|
||||
component,
|
||||
Component::ParentDir | Component::RootDir | Component::Prefix(_)
|
||||
)
|
||||
})
|
||||
{
|
||||
return Err("PERSONA_REMOTE_PATH_INVALID".into());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn validate_head(head: &str) -> Result<&str, String> {
|
||||
if head.len() == 40
|
||||
&& head
|
||||
.bytes()
|
||||
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
|
||||
{
|
||||
Ok(head)
|
||||
} else {
|
||||
Err("PERSONA_REMOTE_HEAD_INVALID".into())
|
||||
}
|
||||
}
|
||||
|
||||
fn prepare_cache_root(cache_root: &Path) -> Result<PathBuf, String> {
|
||||
fs::create_dir_all(cache_root)
|
||||
.map_err(|error| format!("PERSONA_REMOTE_CACHE_ROOT_CREATE_FAILED: {error}"))?;
|
||||
let canonical = cache_root
|
||||
.canonicalize()
|
||||
.map_err(|error| format!("PERSONA_REMOTE_CACHE_ROOT_UNAVAILABLE: {error}"))?;
|
||||
if !canonical.is_dir() {
|
||||
return Err("PERSONA_REMOTE_CACHE_ROOT_NOT_DIRECTORY".into());
|
||||
}
|
||||
Ok(canonical)
|
||||
}
|
||||
|
||||
fn prepare_child_root(storage_root: &Path, child: &str) -> Result<PathBuf, String> {
|
||||
let child_root = storage_root.join(child);
|
||||
fs::create_dir_all(&child_root)
|
||||
.map_err(|error| format!("PERSONA_REMOTE_STORAGE_CHILD_CREATE_FAILED: {error}"))?;
|
||||
let canonical = child_root
|
||||
.canonicalize()
|
||||
.map_err(|error| format!("PERSONA_REMOTE_STORAGE_CHILD_UNAVAILABLE: {error}"))?;
|
||||
if canonical.parent() != Some(storage_root) || !canonical.is_dir() {
|
||||
return Err("PERSONA_REMOTE_STORAGE_BOUNDARY_INVALID".into());
|
||||
}
|
||||
Ok(canonical)
|
||||
}
|
||||
|
||||
fn read_cursor_record(
|
||||
path: &Path,
|
||||
identity_hash: &str,
|
||||
branch: &str,
|
||||
) -> Result<Option<PersonaRemoteCursorRecord>, String> {
|
||||
if !path.exists() {
|
||||
return Ok(None);
|
||||
}
|
||||
let metadata = fs::symlink_metadata(path)
|
||||
.map_err(|error| format!("PERSONA_REMOTE_CURSOR_STATE_UNAVAILABLE: {error}"))?;
|
||||
if metadata.file_type().is_symlink() || !metadata.is_file() {
|
||||
return Err("PERSONA_REMOTE_CURSOR_STATE_INVALID".into());
|
||||
}
|
||||
let bytes = fs::read(path)
|
||||
.map_err(|error| format!("PERSONA_REMOTE_CURSOR_STATE_READ_FAILED: {error}"))?;
|
||||
let record: PersonaRemoteCursorRecord = serde_json::from_slice(&bytes)
|
||||
.map_err(|error| format!("PERSONA_REMOTE_CURSOR_STATE_INVALID: {error}"))?;
|
||||
if record.schema != "hololake.persona-remote-git-cursor/v1"
|
||||
|| record.remote_identity_hash != identity_hash
|
||||
|| record.branch != branch
|
||||
|| record.generation == 0
|
||||
{
|
||||
return Err("PERSONA_REMOTE_CURSOR_STATE_MISMATCH".into());
|
||||
}
|
||||
validate_head(&record.verified_head)?;
|
||||
if let Some(previous) = record.previous_verified_head.as_deref() {
|
||||
validate_head(previous)?;
|
||||
}
|
||||
Ok(Some(record))
|
||||
}
|
||||
|
||||
fn write_cursor_record(path: &Path, record: &PersonaRemoteCursorRecord) -> Result<(), String> {
|
||||
let parent = path
|
||||
.parent()
|
||||
.ok_or_else(|| "PERSONA_REMOTE_CURSOR_STATE_BOUNDARY_INVALID".to_string())?;
|
||||
if !parent.is_dir() || path.parent() != Some(parent) {
|
||||
return Err("PERSONA_REMOTE_CURSOR_STATE_BOUNDARY_INVALID".into());
|
||||
}
|
||||
let bytes = serde_json::to_vec_pretty(record)
|
||||
.map_err(|error| format!("PERSONA_REMOTE_CURSOR_STATE_ENCODE_FAILED: {error}"))?;
|
||||
let temporary = parent.join(format!(".cursor-{}.tmp", Uuid::new_v4()));
|
||||
let mut file = OpenOptions::new()
|
||||
.write(true)
|
||||
.create_new(true)
|
||||
.open(&temporary)
|
||||
.map_err(|error| format!("PERSONA_REMOTE_CURSOR_STATE_CREATE_FAILED: {error}"))?;
|
||||
file.write_all(&bytes)
|
||||
.and_then(|_| file.write_all(b"\n"))
|
||||
.and_then(|_| file.sync_all())
|
||||
.map_err(|error| format!("PERSONA_REMOTE_CURSOR_STATE_WRITE_FAILED: {error}"))?;
|
||||
fs::rename(&temporary, path)
|
||||
.map_err(|error| format!("PERSONA_REMOTE_CURSOR_STATE_COMMIT_FAILED: {error}"))
|
||||
}
|
||||
|
||||
fn cache_key(remote_url: &str, branch: &str) -> String {
|
||||
sha256_hex(format!("{remote_url}\n{branch}").as_bytes())
|
||||
}
|
||||
|
||||
fn sha256_hex(bytes: &[u8]) -> String {
|
||||
digest(&SHA256, bytes)
|
||||
.as_ref()
|
||||
.iter()
|
||||
.map(|byte| format!("{byte:02x}"))
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn read_remote_head(remote_url: &str, branch: &str) -> Result<String, String> {
|
||||
let full_ref = format!("refs/heads/{branch}");
|
||||
let output = Command::new("git")
|
||||
.env("GIT_TRACE_PACKET", "1")
|
||||
.args([
|
||||
"-c",
|
||||
"protocol.version=2",
|
||||
"ls-remote",
|
||||
"--exit-code",
|
||||
"--heads",
|
||||
remote_url,
|
||||
&full_ref,
|
||||
])
|
||||
.output()
|
||||
.map_err(|error| format!("PERSONA_REMOTE_HEAD_READ_FAILED: {error}"))?;
|
||||
let output = require_success("PERSONA_REMOTE_HEAD_READ", output)?;
|
||||
let packet_trace = String::from_utf8_lossy(&output.stderr);
|
||||
if !packet_trace.lines().any(|line| {
|
||||
line.split_whitespace().any(|token| {
|
||||
token.starts_with("fetch=") && line.split_whitespace().any(|part| part == "filter")
|
||||
})
|
||||
}) {
|
||||
return Err("PERSONA_REMOTE_PARTIAL_OBJECT_PROTOCOL_REQUIRED".into());
|
||||
}
|
||||
let output = String::from_utf8(output.stdout)
|
||||
.map_err(|_| "PERSONA_REMOTE_HEAD_READ_NOT_UTF8".to_string())?;
|
||||
let mut lines = output.lines();
|
||||
let first = lines
|
||||
.next()
|
||||
.ok_or_else(|| "PERSONA_REMOTE_BRANCH_NOT_FOUND".to_string())?;
|
||||
if lines.next().is_some() {
|
||||
return Err("PERSONA_REMOTE_HEAD_AMBIGUOUS".into());
|
||||
}
|
||||
let mut fields = first.split_whitespace();
|
||||
let head = fields.next().unwrap_or_default().to_ascii_lowercase();
|
||||
let observed_ref = fields.next().unwrap_or_default();
|
||||
if observed_ref != full_ref || fields.next().is_some() {
|
||||
return Err("PERSONA_REMOTE_HEAD_RESPONSE_INVALID".into());
|
||||
}
|
||||
validate_head(&head)?;
|
||||
Ok(head)
|
||||
}
|
||||
|
||||
fn ensure_bare_partial_cache(cache_path: &Path, remote_url: &str) -> Result<(), String> {
|
||||
if cache_path.exists() {
|
||||
let bare = git_output_text(
|
||||
git_at(cache_path).args(["rev-parse", "--is-bare-repository"]),
|
||||
"PERSONA_REMOTE_CACHE_PROBE",
|
||||
)?;
|
||||
if bare.trim() != "true" {
|
||||
return Err("PERSONA_REMOTE_CACHE_NOT_BARE".into());
|
||||
}
|
||||
let configured = git_output_text(
|
||||
git_at(cache_path).args(["remote", "get-url", "origin"]),
|
||||
"PERSONA_REMOTE_CACHE_ORIGIN_READ",
|
||||
)?;
|
||||
if configured.trim() != remote_url {
|
||||
return Err("PERSONA_REMOTE_CACHE_ORIGIN_MISMATCH".into());
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
fs::create_dir(cache_path)
|
||||
.map_err(|error| format!("PERSONA_REMOTE_CACHE_CREATE_FAILED: {error}"))?;
|
||||
git_status(
|
||||
Command::new("git").args(["init", "--bare", &cache_path.to_string_lossy()]),
|
||||
"PERSONA_REMOTE_CACHE_INIT",
|
||||
)?;
|
||||
git_status(
|
||||
git_at(cache_path).args(["remote", "add", "origin", remote_url]),
|
||||
"PERSONA_REMOTE_CACHE_ORIGIN_ADD",
|
||||
)?;
|
||||
git_status(
|
||||
git_at(cache_path).args(["config", "remote.origin.promisor", "true"]),
|
||||
"PERSONA_REMOTE_CACHE_PROMISOR_CONFIG",
|
||||
)?;
|
||||
git_status(
|
||||
git_at(cache_path).args(["config", "remote.origin.partialclonefilter", "blob:none"]),
|
||||
"PERSONA_REMOTE_CACHE_FILTER_CONFIG",
|
||||
)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn fetch_remote_cursor(
|
||||
cache_path: &Path,
|
||||
branch: &str,
|
||||
previous_cursor: Option<&str>,
|
||||
cache_has_previous_cursor: bool,
|
||||
) -> Result<(), String> {
|
||||
let refspec = format!("+refs/heads/{branch}:refs/remotes/origin/{branch}");
|
||||
if previous_cursor.is_none() || !cache_has_previous_cursor {
|
||||
fetch_at_depth(cache_path, &refspec, 1)?;
|
||||
} else {
|
||||
fetch_without_history_reset(cache_path, &refspec)?;
|
||||
}
|
||||
|
||||
let Some(previous) = previous_cursor else {
|
||||
return Ok(());
|
||||
};
|
||||
let current_ref = format!("refs/remotes/origin/{branch}");
|
||||
let current = required_ref(cache_path, ¤t_ref)?;
|
||||
if previous == current || git_is_ancestor(cache_path, previous, ¤t)? {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
if cache_has_previous_cursor {
|
||||
return Err("PERSONA_REMOTE_HISTORY_REWRITE_REJECTED".into());
|
||||
}
|
||||
|
||||
let mut depth = 2_usize;
|
||||
while depth <= MAX_CONTINUITY_REHYDRATION_DEPTH {
|
||||
fetch_at_depth(cache_path, &refspec, depth)?;
|
||||
if git_is_ancestor(cache_path, previous, ¤t)? {
|
||||
return Ok(());
|
||||
}
|
||||
depth = depth.saturating_mul(2);
|
||||
}
|
||||
Err("PERSONA_REMOTE_CONTINUITY_WINDOW_EXCEEDED".into())
|
||||
}
|
||||
|
||||
fn fetch_at_depth(cache_path: &Path, refspec: &str, depth: usize) -> Result<(), String> {
|
||||
let depth = format!("--depth={depth}");
|
||||
let output = git_at(cache_path)
|
||||
.args([
|
||||
"fetch",
|
||||
"--no-tags",
|
||||
&depth,
|
||||
"--filter=blob:none",
|
||||
"origin",
|
||||
refspec,
|
||||
])
|
||||
.output()
|
||||
.map_err(|error| format!("PERSONA_REMOTE_INCREMENTAL_FETCH_FAILED: {error}"))?;
|
||||
require_partial_fetch_success(output)
|
||||
}
|
||||
|
||||
fn fetch_without_history_reset(cache_path: &Path, refspec: &str) -> Result<(), String> {
|
||||
let output = git_at(cache_path)
|
||||
.args([
|
||||
"fetch",
|
||||
"--no-tags",
|
||||
"--update-shallow",
|
||||
"--filter=blob:none",
|
||||
"origin",
|
||||
refspec,
|
||||
])
|
||||
.output()
|
||||
.map_err(|error| format!("PERSONA_REMOTE_INCREMENTAL_FETCH_FAILED: {error}"))?;
|
||||
require_partial_fetch_success(output)
|
||||
}
|
||||
|
||||
fn require_partial_fetch_success(output: Output) -> Result<(), String> {
|
||||
let output = require_success("PERSONA_REMOTE_INCREMENTAL_FETCH", output)?;
|
||||
let stderr = String::from_utf8_lossy(&output.stderr);
|
||||
if stderr.contains("filtering not recognized") || stderr.contains("filter-spec") {
|
||||
return Err("PERSONA_REMOTE_PARTIAL_OBJECT_PROTOCOL_REQUIRED".into());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn git_is_ancestor(cache_path: &Path, ancestor: &str, descendant: &str) -> Result<bool, String> {
|
||||
validate_head(ancestor)?;
|
||||
validate_head(descendant)?;
|
||||
let output = git_at(cache_path)
|
||||
.args(["merge-base", "--is-ancestor", ancestor, descendant])
|
||||
.output()
|
||||
.map_err(|error| format!("PERSONA_REMOTE_CONTINUITY_CHECK_FAILED: {error}"))?;
|
||||
match output.status.code() {
|
||||
Some(0) => Ok(true),
|
||||
Some(1) => Ok(false),
|
||||
_ => Err(format!(
|
||||
"PERSONA_REMOTE_CONTINUITY_CHECK_FAILED: {}",
|
||||
String::from_utf8_lossy(&output.stderr).trim()
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
fn optional_ref(cache_path: &Path, git_ref: &str) -> Result<Option<String>, String> {
|
||||
let output = git_at(cache_path)
|
||||
.args(["rev-parse", "--verify", git_ref])
|
||||
.output()
|
||||
.map_err(|error| format!("PERSONA_REMOTE_CURSOR_READ_FAILED: {error}"))?;
|
||||
if !output.status.success() {
|
||||
return Ok(None);
|
||||
}
|
||||
let head = String::from_utf8(output.stdout)
|
||||
.map_err(|_| "PERSONA_REMOTE_CURSOR_NOT_UTF8".to_string())?
|
||||
.trim()
|
||||
.to_ascii_lowercase();
|
||||
validate_head(&head)?;
|
||||
Ok(Some(head))
|
||||
}
|
||||
|
||||
fn required_ref(cache_path: &Path, git_ref: &str) -> Result<String, String> {
|
||||
optional_ref(cache_path, git_ref)?
|
||||
.ok_or_else(|| "PERSONA_REMOTE_FETCH_CURSOR_MISSING".to_string())
|
||||
}
|
||||
|
||||
fn git_at(path: &Path) -> Command {
|
||||
let mut command = Command::new("git");
|
||||
command.arg("-C").arg(path);
|
||||
command
|
||||
}
|
||||
|
||||
fn git_status(command: &mut Command, label: &str) -> Result<(), String> {
|
||||
let output = command
|
||||
.output()
|
||||
.map_err(|error| format!("{label}_FAILED: {error}"))?;
|
||||
require_success(label, output).map(|_| ())
|
||||
}
|
||||
|
||||
fn git_output_text(command: &mut Command, label: &str) -> Result<String, String> {
|
||||
let output = command
|
||||
.output()
|
||||
.map_err(|error| format!("{label}_FAILED: {error}"))?;
|
||||
let output = require_success(label, output)?;
|
||||
String::from_utf8(output.stdout).map_err(|_| format!("{label}_NOT_UTF8"))
|
||||
}
|
||||
|
||||
fn git_output_bytes(command: &mut Command, label: &str) -> Result<Vec<u8>, String> {
|
||||
let output = command
|
||||
.output()
|
||||
.map_err(|error| format!("{label}_FAILED: {error}"))?;
|
||||
Ok(require_success(label, output)?.stdout)
|
||||
}
|
||||
|
||||
fn require_success(label: &str, output: Output) -> Result<Output, String> {
|
||||
if output.status.success() {
|
||||
return Ok(output);
|
||||
}
|
||||
let stderr = String::from_utf8_lossy(&output.stderr);
|
||||
Err(format!("{label}_FAILED: {}", stderr.trim()))
|
||||
}
|
||||
|
||||
fn directory_size(root: &Path) -> Result<u64, String> {
|
||||
let mut total = 0_u64;
|
||||
let mut pending = vec![root.to_path_buf()];
|
||||
while let Some(path) = pending.pop() {
|
||||
let metadata = fs::symlink_metadata(&path)
|
||||
.map_err(|error| format!("PERSONA_REMOTE_CACHE_MEASURE_FAILED: {error}"))?;
|
||||
if metadata.file_type().is_symlink() {
|
||||
return Err("PERSONA_REMOTE_CACHE_SYMLINK_REJECTED".into());
|
||||
}
|
||||
if metadata.is_file() {
|
||||
total = total.saturating_add(metadata.len());
|
||||
continue;
|
||||
}
|
||||
if metadata.is_dir() {
|
||||
for entry in fs::read_dir(&path)
|
||||
.map_err(|error| format!("PERSONA_REMOTE_CACHE_MEASURE_FAILED: {error}"))?
|
||||
{
|
||||
pending.push(
|
||||
entry
|
||||
.map_err(|error| format!("PERSONA_REMOTE_CACHE_MEASURE_FAILED: {error}"))?
|
||||
.path(),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(total)
|
||||
}
|
||||
|
||||
fn remove_exact_cache(cache_root: &Path, cache_path: &Path) -> Result<(), String> {
|
||||
let parent = cache_path
|
||||
.parent()
|
||||
.ok_or_else(|| "PERSONA_REMOTE_CACHE_BOUNDARY_INVALID".to_string())?;
|
||||
if parent != cache_root || !cache_path.starts_with(cache_root) || cache_path == cache_root {
|
||||
return Err("PERSONA_REMOTE_CACHE_BOUNDARY_INVALID".into());
|
||||
}
|
||||
fs::remove_dir_all(cache_path)
|
||||
.map_err(|error| format!("PERSONA_REMOTE_CACHE_EVICTION_FAILED: {error}"))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use tempfile::TempDir;
|
||||
|
||||
fn run_git(path: &Path, args: &[&str]) {
|
||||
let output = Command::new("git")
|
||||
.arg("-C")
|
||||
.arg(path)
|
||||
.args(args)
|
||||
.output()
|
||||
.expect("git should run");
|
||||
assert!(
|
||||
output.status.success(),
|
||||
"git {:?} failed: {}",
|
||||
args,
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
}
|
||||
|
||||
fn fixture() -> (TempDir, PathBuf, PathBuf, String) {
|
||||
let temp = TempDir::new().unwrap();
|
||||
let source = temp.path().join("source");
|
||||
let remote = temp.path().join("remote.git");
|
||||
let cache = temp.path().join("cache");
|
||||
fs::create_dir(&source).unwrap();
|
||||
run_git(&source, &["init", "-b", "main"]);
|
||||
run_git(&source, &["config", "user.name", "Persona Test"]);
|
||||
run_git(&source, &["config", "user.email", "persona@test.invalid"]);
|
||||
fs::create_dir_all(source.join("brain")).unwrap();
|
||||
fs::write(source.join("brain/current.hldp"), "first\n").unwrap();
|
||||
run_git(&source, &["add", "."]);
|
||||
run_git(&source, &["commit", "-m", "first"]);
|
||||
let init = Command::new("git")
|
||||
.args(["init", "--bare", &remote.to_string_lossy()])
|
||||
.output()
|
||||
.unwrap();
|
||||
assert!(init.status.success());
|
||||
run_git(&remote, &["config", "uploadpack.allowFilter", "true"]);
|
||||
run_git(
|
||||
&remote,
|
||||
&["config", "uploadpack.allowAnySHA1InWant", "true"],
|
||||
);
|
||||
run_git(
|
||||
&source,
|
||||
&["remote", "add", "origin", &remote.to_string_lossy()],
|
||||
);
|
||||
run_git(&source, &["push", "-u", "origin", "main"]);
|
||||
let remote_url = format!("file://{}", remote.display());
|
||||
(temp, source, cache, remote_url)
|
||||
}
|
||||
|
||||
fn request(cache: &Path, remote_url: &str) -> RemoteReadRequest {
|
||||
RemoteReadRequest {
|
||||
remote_url: remote_url.to_string(),
|
||||
branch: "main".into(),
|
||||
relative_path: "brain/current.hldp".into(),
|
||||
expected_head: None,
|
||||
cache_root: prepare_cache_root(cache).unwrap(),
|
||||
max_cache_bytes: DEFAULT_CACHE_LIMIT_BYTES,
|
||||
allow_file_remote: true,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reads_remote_persona_text_without_a_clone_or_worktree() {
|
||||
let (_temp, _source, cache, remote_url) = fixture();
|
||||
let receipt = read_remote_text_object(&request(&cache, &remote_url)).unwrap();
|
||||
assert_eq!(receipt.content, "first\n");
|
||||
assert!(receipt.cursor_advanced);
|
||||
assert!(receipt.continuity_verified);
|
||||
assert!(receipt.fetched_incremental_objects);
|
||||
assert!(!receipt.cache_rehydrated);
|
||||
assert!(!receipt.worktree_created);
|
||||
assert!(!receipt.full_history_requested);
|
||||
assert_eq!(receipt.cache_state, "BOUNDED_PARTIAL_OBJECT_CACHE");
|
||||
assert!(Path::new(&receipt.cache_path).join("HEAD").is_file());
|
||||
assert!(!Path::new(&receipt.cache_path).join(".git").exists());
|
||||
assert!(Path::new(&receipt.continuity_state_path).is_file());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unchanged_remote_cursor_does_not_fetch_again() {
|
||||
let (_temp, _source, cache, remote_url) = fixture();
|
||||
let first = read_remote_text_object(&request(&cache, &remote_url)).unwrap();
|
||||
let second = read_remote_text_object(&request(&cache, &remote_url)).unwrap();
|
||||
assert_eq!(second.remote_head, first.remote_head);
|
||||
assert_eq!(second.previous_cursor, Some(first.remote_head));
|
||||
assert!(!second.cursor_advanced);
|
||||
assert!(!second.fetched_incremental_objects);
|
||||
assert!(!second.cache_rehydrated);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn changed_remote_cursor_appends_only_the_new_segment_to_existing_continuity() {
|
||||
let (_temp, source, cache, remote_url) = fixture();
|
||||
let first = read_remote_text_object(&request(&cache, &remote_url)).unwrap();
|
||||
fs::write(source.join("brain/current.hldp"), "second\n").unwrap();
|
||||
run_git(&source, &["add", "."]);
|
||||
run_git(&source, &["commit", "-m", "second"]);
|
||||
run_git(&source, &["push", "origin", "main"]);
|
||||
let second = read_remote_text_object(&request(&cache, &remote_url)).unwrap();
|
||||
assert_ne!(second.remote_head, first.remote_head);
|
||||
assert_eq!(second.previous_cursor, Some(first.remote_head));
|
||||
assert!(second.cursor_advanced);
|
||||
assert!(second.fetched_incremental_objects);
|
||||
assert_eq!(second.content, "second\n");
|
||||
let visible_commits = git_output_text(
|
||||
git_at(Path::new(&second.cache_path)).args([
|
||||
"rev-list",
|
||||
"--count",
|
||||
"refs/remotes/origin/main",
|
||||
]),
|
||||
"TEST_VISIBLE_COMMIT_COUNT",
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(visible_commits.trim(), "2");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn oversized_partial_cache_is_evicted_after_the_bounded_read() {
|
||||
let (_temp, _source, cache, remote_url) = fixture();
|
||||
let mut input = request(&cache, &remote_url);
|
||||
input.max_cache_bytes = 1;
|
||||
let receipt = read_remote_text_object(&input).unwrap();
|
||||
assert_eq!(receipt.content, "first\n");
|
||||
assert_eq!(receipt.cache_state, "EVICTED_AFTER_BOUNDED_READ");
|
||||
assert_eq!(receipt.cache_bytes, 0);
|
||||
assert!(!Path::new(&receipt.cache_path).exists());
|
||||
assert!(Path::new(&receipt.continuity_state_path).is_file());
|
||||
let state: PersonaRemoteCursorRecord =
|
||||
serde_json::from_slice(&fs::read(&receipt.continuity_state_path).unwrap()).unwrap();
|
||||
assert_eq!(state.verified_head, receipt.remote_head);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn evicted_object_cache_rehydrates_only_the_missing_new_segments() {
|
||||
let (_temp, source, storage, remote_url) = fixture();
|
||||
let mut first_request = request(&storage, &remote_url);
|
||||
first_request.max_cache_bytes = 1;
|
||||
let first = read_remote_text_object(&first_request).unwrap();
|
||||
assert!(!Path::new(&first.cache_path).exists());
|
||||
|
||||
for (message, content) in [
|
||||
("second", "second\n"),
|
||||
("third", "third\n"),
|
||||
("fourth", "fourth\n"),
|
||||
] {
|
||||
fs::write(source.join("brain/current.hldp"), content).unwrap();
|
||||
run_git(&source, &["add", "."]);
|
||||
run_git(&source, &["commit", "-m", message]);
|
||||
}
|
||||
run_git(&source, &["push", "origin", "main"]);
|
||||
|
||||
let second = read_remote_text_object(&request(&storage, &remote_url)).unwrap();
|
||||
assert_eq!(second.previous_cursor, Some(first.remote_head));
|
||||
assert!(second.cursor_advanced);
|
||||
assert!(second.cache_rehydrated);
|
||||
assert!(second.continuity_verified);
|
||||
assert_eq!(second.content, "fourth\n");
|
||||
let visible_commits = git_output_text(
|
||||
git_at(Path::new(&second.cache_path)).args([
|
||||
"rev-list",
|
||||
"--count",
|
||||
"refs/remotes/origin/main",
|
||||
]),
|
||||
"TEST_REHYDRATED_COMMIT_COUNT",
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(visible_commits.trim(), "4");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_remote_history_rewrite_without_advancing_the_durable_cursor() {
|
||||
let (_temp, source, storage, remote_url) = fixture();
|
||||
let first = read_remote_text_object(&request(&storage, &remote_url)).unwrap();
|
||||
run_git(&source, &["checkout", "--orphan", "rewritten"]);
|
||||
run_git(&source, &["rm", "-rf", "."]);
|
||||
fs::create_dir_all(source.join("brain")).unwrap();
|
||||
fs::write(source.join("brain/current.hldp"), "rewritten\n").unwrap();
|
||||
run_git(&source, &["add", "."]);
|
||||
run_git(&source, &["commit", "-m", "rewrite"]);
|
||||
run_git(&source, &["branch", "-M", "main"]);
|
||||
run_git(&source, &["push", "--force", "origin", "main"]);
|
||||
|
||||
assert_eq!(
|
||||
read_remote_text_object(&request(&storage, &remote_url)).unwrap_err(),
|
||||
"PERSONA_REMOTE_HISTORY_REWRITE_REJECTED"
|
||||
);
|
||||
let state: PersonaRemoteCursorRecord =
|
||||
serde_json::from_slice(&fs::read(&first.continuity_state_path).unwrap()).unwrap();
|
||||
assert_eq!(state.verified_head, first.remote_head);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_credentials_path_escape_and_wrong_expected_head() {
|
||||
assert_eq!(
|
||||
validate_remote_url("https://token@example.invalid/repo.git", false).unwrap_err(),
|
||||
"PERSONA_REMOTE_URL_MUST_NOT_CONTAIN_CREDENTIALS"
|
||||
);
|
||||
assert_eq!(
|
||||
validate_relative_path("../brain.hldp").unwrap_err(),
|
||||
"PERSONA_REMOTE_PATH_INVALID"
|
||||
);
|
||||
let (_temp, _source, cache, remote_url) = fixture();
|
||||
let mut input = request(&cache, &remote_url);
|
||||
input.expected_head = Some("0000000000000000000000000000000000000000".into());
|
||||
assert_eq!(
|
||||
read_remote_text_object(&input).unwrap_err(),
|
||||
"PERSONA_REMOTE_HEAD_MISMATCH"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_a_remote_that_cannot_serve_partial_objects_before_fetching() {
|
||||
let (_temp, _source, storage, remote_url) = fixture();
|
||||
let remote_path = PathBuf::from(remote_url.trim_start_matches("file://"));
|
||||
run_git(&remote_path, &["config", "uploadpack.allowFilter", "false"]);
|
||||
assert_eq!(
|
||||
read_remote_text_object(&request(&storage, &remote_url)).unwrap_err(),
|
||||
"PERSONA_REMOTE_PARTIAL_OBJECT_PROTOCOL_REQUIRED"
|
||||
);
|
||||
assert!(!storage.join("objects").exists());
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue