Implement authorized landmark roaming (#130)
This commit is contained in:
@@ -187,6 +187,14 @@ impl LibremetaverseClientOwner {
|
|||||||
&self.client
|
&self.client
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Returns the one native agent manager owned by this composition root.
|
||||||
|
/// `MetaCrate` adapters clone the `Arc`; they never construct a parallel
|
||||||
|
/// compatibility manager graph.
|
||||||
|
#[must_use]
|
||||||
|
pub fn agent(&self) -> Arc<libremetaverse::AgentManager> {
|
||||||
|
Arc::clone(&self.agent)
|
||||||
|
}
|
||||||
|
|
||||||
/// Creates the supervised live-session adapter without logging in.
|
/// Creates the supervised live-session adapter without logging in.
|
||||||
///
|
///
|
||||||
/// # Errors
|
/// # Errors
|
||||||
|
|||||||
557
crates/metacrate-grid-agent/src/landmark_tests.rs
Normal file
557
crates/metacrate-grid-agent/src/landmark_tests.rs
Normal file
@@ -0,0 +1,557 @@
|
|||||||
|
use crate::ProposedToolCall;
|
||||||
|
use crate::landmarks::*;
|
||||||
|
use crate::policy::{
|
||||||
|
ActionOrigin, MemoryPolicyAudit, PolicyGateway, PolicyLimits, PolicyRequestContext,
|
||||||
|
};
|
||||||
|
use libremetaverse_types::compat::CancellationTokenSource;
|
||||||
|
use libremetaverse_types::{AssetType, UUID, compat::CancellationToken};
|
||||||
|
use std::collections::BTreeSet;
|
||||||
|
use std::sync::{Arc, Mutex};
|
||||||
|
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||||
|
|
||||||
|
fn id(value: u128) -> UUID {
|
||||||
|
UUID::new_with_string(format!("{value:032x}")).expect("uuid")
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Default)]
|
||||||
|
struct FakeGrid {
|
||||||
|
accepted: Mutex<Vec<String>>,
|
||||||
|
declined: Mutex<Vec<String>>,
|
||||||
|
teleports: Mutex<Vec<UUID>>,
|
||||||
|
valid: Mutex<bool>,
|
||||||
|
succeed: Mutex<bool>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl LandmarkGrid for FakeGrid {
|
||||||
|
fn accept_offer(&self, offer_id: &str, _: CancellationToken) -> LandmarkFuture<'_, ()> {
|
||||||
|
let offer_id = offer_id.to_owned();
|
||||||
|
Box::pin(async move {
|
||||||
|
self.accepted.lock().expect("accepted").push(offer_id);
|
||||||
|
Ok(())
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
fn decline_offer(&self, offer_id: &str, _: CancellationToken) -> LandmarkFuture<'_, ()> {
|
||||||
|
let offer_id = offer_id.to_owned();
|
||||||
|
Box::pin(async move {
|
||||||
|
self.declined.lock().expect("declined").push(offer_id);
|
||||||
|
Ok(())
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
fn verify_landmark(
|
||||||
|
&self,
|
||||||
|
_: UUID,
|
||||||
|
_: UUID,
|
||||||
|
_: u64,
|
||||||
|
_: CancellationToken,
|
||||||
|
) -> LandmarkFuture<'_, bool> {
|
||||||
|
Box::pin(async move { Ok(*self.valid.lock().expect("valid")) })
|
||||||
|
}
|
||||||
|
|
||||||
|
fn teleport_landmark(&self, asset_id: UUID, _: CancellationToken) -> LandmarkFuture<'_, bool> {
|
||||||
|
Box::pin(async move {
|
||||||
|
self.teleports.lock().expect("teleports").push(asset_id);
|
||||||
|
Ok(*self.succeed.lock().expect("succeed"))
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug)]
|
||||||
|
struct FixedRandom;
|
||||||
|
impl RoamingRandom for FixedRandom {
|
||||||
|
fn index(&self, upper_exclusive: usize) -> usize {
|
||||||
|
upper_exclusive.saturating_sub(1)
|
||||||
|
}
|
||||||
|
fn interval_seconds(&self, minimum: u64, _: u64) -> u64 {
|
||||||
|
minimum
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn landmark(inventory: u128, asset: u128, owner: u128, name: &str) -> OfferedInventoryNode {
|
||||||
|
OfferedInventoryNode {
|
||||||
|
inventory_id: id(inventory),
|
||||||
|
owner_id: id(owner),
|
||||||
|
name: name.into(),
|
||||||
|
kind: OfferedInventoryKind::Landmark {
|
||||||
|
asset_id: id(asset),
|
||||||
|
permissions_fingerprint: 0xCAFE,
|
||||||
|
},
|
||||||
|
children: Vec::new(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn offer(sender: u128, root: OfferedInventoryNode, offer_id: &str) -> LandmarkOffer {
|
||||||
|
LandmarkOffer {
|
||||||
|
offer_id: offer_id.into(),
|
||||||
|
sender_id: id(sender),
|
||||||
|
root,
|
||||||
|
received_unix_millis: 1_000,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn service(grid: Arc<FakeGrid>, authorized: &[u128]) -> LandmarkService {
|
||||||
|
*grid.valid.lock().expect("valid") = true;
|
||||||
|
*grid.succeed.lock().expect("succeed") = true;
|
||||||
|
LandmarkService::new(
|
||||||
|
grid,
|
||||||
|
Arc::new(FixedRandom),
|
||||||
|
authorized.iter().copied().map(id).collect(),
|
||||||
|
LandmarkLimits {
|
||||||
|
teleport_cooldown: Duration::ZERO,
|
||||||
|
..LandmarkLimits::default()
|
||||||
|
},
|
||||||
|
None,
|
||||||
|
)
|
||||||
|
.expect("service")
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn authorized_nested_landmarks_accept_once_and_unauthorized_or_bad_types_decline() {
|
||||||
|
let grid = Arc::new(FakeGrid::default());
|
||||||
|
let service = service(grid.clone(), &[10]);
|
||||||
|
let nested = OfferedInventoryNode {
|
||||||
|
inventory_id: id(100),
|
||||||
|
owner_id: id(10),
|
||||||
|
name: "Trips".into(),
|
||||||
|
kind: OfferedInventoryKind::Folder,
|
||||||
|
children: vec![
|
||||||
|
landmark(101, 201, 10, "Alpha"),
|
||||||
|
landmark(102, 202, 10, "Beta 世界"),
|
||||||
|
],
|
||||||
|
};
|
||||||
|
assert_eq!(
|
||||||
|
service
|
||||||
|
.ingest_offer(
|
||||||
|
offer(10, nested.clone(), "offer-1"),
|
||||||
|
CancellationToken::default()
|
||||||
|
)
|
||||||
|
.await,
|
||||||
|
Ok(OfferDecision::Accepted)
|
||||||
|
);
|
||||||
|
assert_eq!(service.entries().len(), 2);
|
||||||
|
assert_eq!(
|
||||||
|
service
|
||||||
|
.ingest_offer(offer(10, nested, "offer-1"), CancellationToken::default())
|
||||||
|
.await,
|
||||||
|
Ok(OfferDecision::Duplicate)
|
||||||
|
);
|
||||||
|
assert_eq!(grid.accepted.lock().expect("accepted").len(), 1);
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
service
|
||||||
|
.ingest_offer(
|
||||||
|
offer(11, landmark(103, 203, 11, "No"), "offer-2"),
|
||||||
|
CancellationToken::default(),
|
||||||
|
)
|
||||||
|
.await,
|
||||||
|
Err(LandmarkError::Unauthorized)
|
||||||
|
);
|
||||||
|
let bad = OfferedInventoryNode {
|
||||||
|
inventory_id: id(104),
|
||||||
|
owner_id: id(10),
|
||||||
|
name: "Money".into(),
|
||||||
|
kind: OfferedInventoryKind::Other(AssetType::CallingCard),
|
||||||
|
children: Vec::new(),
|
||||||
|
};
|
||||||
|
assert_eq!(
|
||||||
|
service
|
||||||
|
.ingest_offer(offer(10, bad, "offer-3"), CancellationToken::default())
|
||||||
|
.await,
|
||||||
|
Err(LandmarkError::UnsupportedInventory)
|
||||||
|
);
|
||||||
|
assert_eq!(grid.declined.lock().expect("declined").len(), 2);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn cycles_duplicate_assets_and_ambiguous_names_fail_closed() {
|
||||||
|
let grid = Arc::new(FakeGrid::default());
|
||||||
|
let service = service(grid, &[20]);
|
||||||
|
let duplicate = OfferedInventoryNode {
|
||||||
|
inventory_id: id(300),
|
||||||
|
owner_id: id(20),
|
||||||
|
name: "Dupes".into(),
|
||||||
|
kind: OfferedInventoryKind::Folder,
|
||||||
|
children: vec![
|
||||||
|
landmark(301, 401, 20, "Same"),
|
||||||
|
landmark(302, 401, 20, "Same"),
|
||||||
|
],
|
||||||
|
};
|
||||||
|
assert_eq!(
|
||||||
|
service
|
||||||
|
.ingest_offer(offer(20, duplicate, "dup"), CancellationToken::default())
|
||||||
|
.await,
|
||||||
|
Err(LandmarkError::InvalidOffer)
|
||||||
|
);
|
||||||
|
|
||||||
|
let folder = OfferedInventoryNode {
|
||||||
|
inventory_id: id(310),
|
||||||
|
owner_id: id(20),
|
||||||
|
name: "Names".into(),
|
||||||
|
kind: OfferedInventoryKind::Folder,
|
||||||
|
children: vec![
|
||||||
|
landmark(311, 411, 20, "Same"),
|
||||||
|
landmark(312, 412, 20, "Same"),
|
||||||
|
],
|
||||||
|
};
|
||||||
|
service
|
||||||
|
.ingest_offer(offer(20, folder, "names"), CancellationToken::default())
|
||||||
|
.await
|
||||||
|
.expect("accepted");
|
||||||
|
assert_eq!(
|
||||||
|
service.select("Same"),
|
||||||
|
Err(LandmarkError::AmbiguousSelection)
|
||||||
|
);
|
||||||
|
assert!(service.select(&format!("lm-{}", id(311))).is_ok());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn teleport_revalidates_stable_inventory_and_never_uses_message_destination() {
|
||||||
|
let grid = Arc::new(FakeGrid::default());
|
||||||
|
let service = service(grid.clone(), &[30]);
|
||||||
|
service
|
||||||
|
.ingest_offer(
|
||||||
|
offer(30, landmark(501, 601, 30, "Home"), "home"),
|
||||||
|
CancellationToken::default(),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("offer");
|
||||||
|
let receipt = service
|
||||||
|
.teleport(
|
||||||
|
id(30),
|
||||||
|
"Home",
|
||||||
|
"command-1",
|
||||||
|
TeleportTrigger::Command,
|
||||||
|
CancellationToken::default(),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("teleport");
|
||||||
|
assert!(receipt.succeeded);
|
||||||
|
assert_eq!(service.observations().len(), 2);
|
||||||
|
assert_eq!(
|
||||||
|
service.observations()[1].kind,
|
||||||
|
LandmarkObservationKind::TeleportSucceeded
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
grid.teleports.lock().expect("teleports").as_slice(),
|
||||||
|
&[id(601)]
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
service
|
||||||
|
.teleport(
|
||||||
|
id(31),
|
||||||
|
&format!("lm-{}", id(501)),
|
||||||
|
"spoof",
|
||||||
|
TeleportTrigger::Command,
|
||||||
|
CancellationToken::default(),
|
||||||
|
)
|
||||||
|
.await,
|
||||||
|
Err(LandmarkError::Unauthorized)
|
||||||
|
);
|
||||||
|
*grid.valid.lock().expect("valid") = false;
|
||||||
|
assert_eq!(
|
||||||
|
service
|
||||||
|
.teleport(
|
||||||
|
id(30),
|
||||||
|
"Home",
|
||||||
|
"stale",
|
||||||
|
TeleportTrigger::Command,
|
||||||
|
CancellationToken::default(),
|
||||||
|
)
|
||||||
|
.await,
|
||||||
|
Err(LandmarkError::StaleLandmark)
|
||||||
|
);
|
||||||
|
assert_eq!(grid.teleports.lock().expect("teleports").len(), 1);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn cancellation_failure_and_cooldown_are_bounded_without_duplicate_teleports() {
|
||||||
|
let grid = Arc::new(FakeGrid::default());
|
||||||
|
*grid.valid.lock().expect("valid") = true;
|
||||||
|
*grid.succeed.lock().expect("succeed") = false;
|
||||||
|
let service = LandmarkService::new(
|
||||||
|
grid.clone(),
|
||||||
|
Arc::new(FixedRandom),
|
||||||
|
BTreeSet::from([id(35)]),
|
||||||
|
LandmarkLimits {
|
||||||
|
teleport_cooldown: Duration::from_secs(30),
|
||||||
|
..LandmarkLimits::default()
|
||||||
|
},
|
||||||
|
None,
|
||||||
|
)
|
||||||
|
.expect("service");
|
||||||
|
service
|
||||||
|
.ingest_offer(
|
||||||
|
offer(35, landmark(551, 651, 35, "Failure"), "failure"),
|
||||||
|
CancellationToken::default(),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("offer");
|
||||||
|
let cancelled = CancellationTokenSource::new();
|
||||||
|
cancelled.cancel();
|
||||||
|
assert_eq!(
|
||||||
|
service
|
||||||
|
.teleport(
|
||||||
|
id(35),
|
||||||
|
"Failure",
|
||||||
|
"cancelled",
|
||||||
|
TeleportTrigger::Command,
|
||||||
|
cancelled.token(),
|
||||||
|
)
|
||||||
|
.await,
|
||||||
|
Err(LandmarkError::Cancelled)
|
||||||
|
);
|
||||||
|
assert!(grid.teleports.lock().expect("teleports").is_empty());
|
||||||
|
assert_eq!(
|
||||||
|
service
|
||||||
|
.teleport(
|
||||||
|
id(35),
|
||||||
|
"Failure",
|
||||||
|
"failed",
|
||||||
|
TeleportTrigger::Command,
|
||||||
|
CancellationToken::default(),
|
||||||
|
)
|
||||||
|
.await,
|
||||||
|
Err(LandmarkError::TeleportFailed)
|
||||||
|
);
|
||||||
|
// A failed native attempt still starts the cooldown and cannot be replayed.
|
||||||
|
assert_eq!(
|
||||||
|
service
|
||||||
|
.teleport(
|
||||||
|
id(35),
|
||||||
|
"Failure",
|
||||||
|
"duplicate",
|
||||||
|
TeleportTrigger::Command,
|
||||||
|
CancellationToken::default(),
|
||||||
|
)
|
||||||
|
.await,
|
||||||
|
Err(LandmarkError::Cooldown)
|
||||||
|
);
|
||||||
|
assert_eq!(grid.teleports.lock().expect("teleports").len(), 1);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn roaming_persists_principal_skips_missed_runs_avoids_repeat_and_obeys_pause() {
|
||||||
|
let grid = Arc::new(FakeGrid::default());
|
||||||
|
let service = service(grid, &[40]);
|
||||||
|
for (offer_id, inventory, asset, name) in [("a", 701, 801, "A"), ("b", 702, 802, "B")] {
|
||||||
|
service
|
||||||
|
.ingest_offer(
|
||||||
|
offer(40, landmark(inventory, asset, 40, name), offer_id),
|
||||||
|
CancellationToken::default(),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("offer");
|
||||||
|
}
|
||||||
|
let schedule = service
|
||||||
|
.upsert_schedule(
|
||||||
|
id(40),
|
||||||
|
"daily",
|
||||||
|
Duration::from_mins(5),
|
||||||
|
Duration::from_mins(10),
|
||||||
|
true,
|
||||||
|
1_000,
|
||||||
|
)
|
||||||
|
.expect("schedule");
|
||||||
|
assert_eq!(schedule.authorizing_avatar, id(40).to_string());
|
||||||
|
assert_eq!(
|
||||||
|
service.due_roam_selection(
|
||||||
|
"daily",
|
||||||
|
schedule.next_run_unix_millis,
|
||||||
|
RoamingPause {
|
||||||
|
conversation: true,
|
||||||
|
..RoamingPause::default()
|
||||||
|
},
|
||||||
|
),
|
||||||
|
Err(LandmarkError::Paused)
|
||||||
|
);
|
||||||
|
let first = service
|
||||||
|
.due_roam_selection(
|
||||||
|
"daily",
|
||||||
|
schedule.next_run_unix_millis,
|
||||||
|
RoamingPause::default(),
|
||||||
|
)
|
||||||
|
.expect("due")
|
||||||
|
.expect("selection");
|
||||||
|
let next = service.schedules()[0].next_run_unix_millis;
|
||||||
|
assert!(next > schedule.next_run_unix_millis);
|
||||||
|
let second = service
|
||||||
|
.due_roam_selection("daily", next, RoamingPause::default())
|
||||||
|
.expect("next due")
|
||||||
|
.expect("next selection");
|
||||||
|
assert_ne!(first.1, second.1);
|
||||||
|
assert_eq!(first.0, id(40));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn catalog_and_schedule_survive_restart_without_replaying_a_missed_run() {
|
||||||
|
let directory = std::env::temp_dir().join(format!("metacrate-landmarks-{}", id(999)));
|
||||||
|
std::fs::create_dir_all(&directory).expect("directory");
|
||||||
|
let path = directory.join("catalog.json");
|
||||||
|
let grid = Arc::new(FakeGrid::default());
|
||||||
|
*grid.valid.lock().expect("valid") = true;
|
||||||
|
*grid.succeed.lock().expect("succeed") = true;
|
||||||
|
{
|
||||||
|
let service = LandmarkService::new(
|
||||||
|
grid.clone(),
|
||||||
|
Arc::new(FixedRandom),
|
||||||
|
BTreeSet::from([id(50)]),
|
||||||
|
LandmarkLimits {
|
||||||
|
teleport_cooldown: Duration::ZERO,
|
||||||
|
..LandmarkLimits::default()
|
||||||
|
},
|
||||||
|
Some(path.clone()),
|
||||||
|
)
|
||||||
|
.expect("service");
|
||||||
|
service
|
||||||
|
.ingest_offer(
|
||||||
|
offer(50, landmark(901, 902, 50, "Persisted"), "persist"),
|
||||||
|
CancellationToken::default(),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("offer");
|
||||||
|
service
|
||||||
|
.upsert_schedule(
|
||||||
|
id(50),
|
||||||
|
"restart",
|
||||||
|
Duration::from_mins(5),
|
||||||
|
Duration::from_mins(5),
|
||||||
|
true,
|
||||||
|
10_000,
|
||||||
|
)
|
||||||
|
.expect("schedule");
|
||||||
|
}
|
||||||
|
let restored = LandmarkService::new(
|
||||||
|
grid,
|
||||||
|
Arc::new(FixedRandom),
|
||||||
|
BTreeSet::from([id(50)]),
|
||||||
|
LandmarkLimits {
|
||||||
|
teleport_cooldown: Duration::ZERO,
|
||||||
|
..LandmarkLimits::default()
|
||||||
|
},
|
||||||
|
Some(path.clone()),
|
||||||
|
)
|
||||||
|
.expect("restore");
|
||||||
|
assert_eq!(restored.entries().len(), 1);
|
||||||
|
let before = restored.schedules()[0].next_run_unix_millis;
|
||||||
|
let selection = restored
|
||||||
|
.due_roam_selection("restart", before + 3_600_000, RoamingPause::default())
|
||||||
|
.expect("due")
|
||||||
|
.expect("selection");
|
||||||
|
assert_eq!(selection.0, id(50));
|
||||||
|
assert!(restored.schedules()[0].next_run_unix_millis > before + 3_600_000);
|
||||||
|
let _ = std::fs::remove_file(path);
|
||||||
|
let _ = std::fs::remove_dir(directory);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test(start_paused = true)]
|
||||||
|
async fn owned_runner_executes_due_schedule_once_and_cancels_cleanly() {
|
||||||
|
let grid = Arc::new(FakeGrid::default());
|
||||||
|
let service = Arc::new(service(grid.clone(), &[55]));
|
||||||
|
service
|
||||||
|
.ingest_offer(
|
||||||
|
offer(55, landmark(951, 952, 55, "Runner"), "runner-offer"),
|
||||||
|
CancellationToken::default(),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("offer");
|
||||||
|
let now = u64::try_from(
|
||||||
|
SystemTime::now()
|
||||||
|
.duration_since(UNIX_EPOCH)
|
||||||
|
.expect("clock")
|
||||||
|
.as_millis(),
|
||||||
|
)
|
||||||
|
.expect("millis");
|
||||||
|
service
|
||||||
|
.upsert_schedule(
|
||||||
|
id(55),
|
||||||
|
"runner",
|
||||||
|
Duration::from_mins(5),
|
||||||
|
Duration::from_mins(5),
|
||||||
|
true,
|
||||||
|
now.saturating_sub(301_000),
|
||||||
|
)
|
||||||
|
.expect("schedule");
|
||||||
|
let runner = service.start_roaming_runner(RoamingPause::default(), None);
|
||||||
|
tokio::time::advance(Duration::from_secs(1)).await;
|
||||||
|
for _ in 0..8 {
|
||||||
|
tokio::task::yield_now().await;
|
||||||
|
}
|
||||||
|
assert_eq!(
|
||||||
|
grid.teleports.lock().expect("teleports").as_slice(),
|
||||||
|
&[id(952)]
|
||||||
|
);
|
||||||
|
runner.shutdown().await;
|
||||||
|
tokio::time::advance(Duration::from_hours(1)).await;
|
||||||
|
assert_eq!(grid.teleports.lock().expect("teleports").len(), 1);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn unsafe_bounds_and_empty_authorization_fail_before_side_effects() {
|
||||||
|
let grid = Arc::new(FakeGrid::default());
|
||||||
|
let result = LandmarkService::new(
|
||||||
|
grid,
|
||||||
|
Arc::new(FixedRandom),
|
||||||
|
BTreeSet::new(),
|
||||||
|
LandmarkLimits {
|
||||||
|
max_folder_depth: 0,
|
||||||
|
..LandmarkLimits::default()
|
||||||
|
},
|
||||||
|
None,
|
||||||
|
);
|
||||||
|
assert!(matches!(result, Err(LandmarkError::UnsafeLimits)));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn public_and_unauthorized_im_cannot_select_teleport_or_change_roaming() {
|
||||||
|
let tools = landmark_policy_tools(LandmarkLimits::default()).expect("tools");
|
||||||
|
let gateway = PolicyGateway::new(
|
||||||
|
BTreeSet::from([id(60)]),
|
||||||
|
tools,
|
||||||
|
PolicyLimits::default(),
|
||||||
|
Arc::new(MemoryPolicyAudit::new(64).expect("audit")),
|
||||||
|
)
|
||||||
|
.expect("gateway");
|
||||||
|
for (origin, name, arguments) in [
|
||||||
|
(
|
||||||
|
ActionOrigin::public_chat(id(60)),
|
||||||
|
LANDMARK_TELEPORT_TOOL,
|
||||||
|
serde_json::json!({"selector":"lm-safe"}),
|
||||||
|
),
|
||||||
|
(
|
||||||
|
ActionOrigin::instant_message(id(61)),
|
||||||
|
LANDMARK_SCHEDULE_TOOL,
|
||||||
|
serde_json::json!({
|
||||||
|
"schedule_id":"no",
|
||||||
|
"minimum_interval_seconds":300,
|
||||||
|
"maximum_interval_seconds":600,
|
||||||
|
"enabled":true
|
||||||
|
}),
|
||||||
|
),
|
||||||
|
] {
|
||||||
|
let encoded = arguments.to_string();
|
||||||
|
let context = PolicyRequestContext::new(origin, "session", "correlation").expect("context");
|
||||||
|
let call = ProposedToolCall::new("call", name, encoded).expect("call");
|
||||||
|
let decision = gateway
|
||||||
|
.evaluate(&context, &call, &arguments, None, 1)
|
||||||
|
.expect("decision");
|
||||||
|
assert!(decision.into_authorization().is_none());
|
||||||
|
}
|
||||||
|
|
||||||
|
let arguments = serde_json::json!({"selector":"lm-safe"});
|
||||||
|
let context = PolicyRequestContext::new(
|
||||||
|
ActionOrigin::instant_message(id(60)),
|
||||||
|
"session",
|
||||||
|
"authorized",
|
||||||
|
)
|
||||||
|
.expect("context");
|
||||||
|
let call =
|
||||||
|
ProposedToolCall::new("call", LANDMARK_TELEPORT_TOOL, arguments.to_string()).expect("call");
|
||||||
|
assert!(
|
||||||
|
gateway
|
||||||
|
.evaluate(&context, &call, &arguments, None, 1)
|
||||||
|
.expect("decision")
|
||||||
|
.into_authorization()
|
||||||
|
.is_some()
|
||||||
|
);
|
||||||
|
}
|
||||||
1738
crates/metacrate-grid-agent/src/landmarks.rs
Normal file
1738
crates/metacrate-grid-agent/src/landmarks.rs
Normal file
File diff suppressed because it is too large
Load Diff
@@ -11,6 +11,7 @@ pub mod control_plane;
|
|||||||
pub mod control_runtime;
|
pub mod control_runtime;
|
||||||
pub mod conversation;
|
pub mod conversation;
|
||||||
pub mod interaction;
|
pub mod interaction;
|
||||||
|
pub mod landmarks;
|
||||||
pub mod llm;
|
pub mod llm;
|
||||||
pub mod observability;
|
pub mod observability;
|
||||||
pub mod perception;
|
pub mod perception;
|
||||||
@@ -33,6 +34,8 @@ mod conversation_tests;
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod interaction_tests;
|
mod interaction_tests;
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
|
mod landmark_tests;
|
||||||
|
#[cfg(test)]
|
||||||
mod observability_tests;
|
mod observability_tests;
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod perception_tests;
|
mod perception_tests;
|
||||||
@@ -90,6 +93,18 @@ pub use interaction::{
|
|||||||
PolicyLlmResponder, ResponderFuture, ResponsePacer, ResponseRequest, SuppressionReason,
|
PolicyLlmResponder, ResponderFuture, ResponsePacer, ResponseRequest, SuppressionReason,
|
||||||
VisibleResponse, split_utf8,
|
VisibleResponse, split_utf8,
|
||||||
};
|
};
|
||||||
|
pub use landmarks::{
|
||||||
|
LANDMARK_LIST_TOOL, LANDMARK_SCHEDULE_TOOL, LANDMARK_STATUS_TOOL, LANDMARK_TELEPORT_TOOL,
|
||||||
|
LandmarkEntry, LandmarkError, LandmarkFuture, LandmarkGrid, LandmarkLimits,
|
||||||
|
LandmarkObservation, LandmarkObservationKind, LandmarkOffer, LandmarkOutcome,
|
||||||
|
LandmarkRoamingHandle, LandmarkService, LandmarkToolBackend, LandmarkValidation, OfferDecision,
|
||||||
|
OfferedInventoryKind, OfferedInventoryNode, RoamingPause, RoamingRandom, RoamingSchedule,
|
||||||
|
SystemRoamingRandom, TeleportReceipt, TeleportTrigger, landmark_policy_tools,
|
||||||
|
};
|
||||||
|
#[cfg(feature = "live-grid")]
|
||||||
|
pub use landmarks::{
|
||||||
|
LibremetaverseLandmarkGrid, LibremetaverseLandmarkIntake, permission_fingerprint,
|
||||||
|
};
|
||||||
pub use llm::{
|
pub use llm::{
|
||||||
Completion, CompletionMessage, ContentPart, ImageDetail, LlmClient, LlmError,
|
Completion, CompletionMessage, ContentPart, ImageDetail, LlmClient, LlmError,
|
||||||
LlmTransportLimits, ToolDefinition, ToolSchema, Usage,
|
LlmTransportLimits, ToolDefinition, ToolSchema, Usage,
|
||||||
|
|||||||
@@ -346,6 +346,7 @@ async fn run_live(
|
|||||||
println!("grid agent session supervisor started; press Ctrl-C to stop");
|
println!("grid agent session supervisor started; press Ctrl-C to stop");
|
||||||
let mut signal_error = None;
|
let mut signal_error = None;
|
||||||
let mut control_failure = None;
|
let mut control_failure = None;
|
||||||
|
let mut active_interactions = 0_usize;
|
||||||
loop {
|
loop {
|
||||||
tokio::select! {
|
tokio::select! {
|
||||||
signal = tokio::signal::ctrl_c() => {
|
signal = tokio::signal::ctrl_c() => {
|
||||||
@@ -359,6 +360,8 @@ async fn run_live(
|
|||||||
record_session_observation(&live.observability, &event);
|
record_session_observation(&live.observability, &event);
|
||||||
if let SessionObservation::Transition { status, reason, retry_in } = event {
|
if let SessionObservation::Transition { status, reason, retry_in } = event {
|
||||||
control_target.update_session(status);
|
control_target.update_session(status);
|
||||||
|
live.landmark_roaming
|
||||||
|
.update_pause(|pause| pause.degraded = !status.agent_ready);
|
||||||
control_plane.publish(ControlEventKind::StateChanged {
|
control_plane.publish(ControlEventKind::StateChanged {
|
||||||
component: "session".into(),
|
component: "session".into(),
|
||||||
state: status.state.as_str().into(),
|
state: status.state.as_str().into(),
|
||||||
@@ -374,6 +377,18 @@ async fn run_live(
|
|||||||
}
|
}
|
||||||
event = live.interaction.next_observation() => {
|
event = live.interaction.next_observation() => {
|
||||||
let Some(event) = event else { break; };
|
let Some(event) = event else { break; };
|
||||||
|
match &event {
|
||||||
|
metacrate_grid_agent::InteractionObservation::InferenceStarted { .. } => {
|
||||||
|
active_interactions = active_interactions.saturating_add(1);
|
||||||
|
}
|
||||||
|
metacrate_grid_agent::InteractionObservation::InferenceFinished { .. } => {
|
||||||
|
active_interactions = active_interactions.saturating_sub(1);
|
||||||
|
}
|
||||||
|
_ => {}
|
||||||
|
}
|
||||||
|
live.landmark_roaming.update_pause(|pause| {
|
||||||
|
pause.conversation = active_interactions != 0;
|
||||||
|
});
|
||||||
record_interaction_observation(&live.observability, &event);
|
record_interaction_observation(&live.observability, &event);
|
||||||
println!("grid interaction event={event:?}");
|
println!("grid interaction event={event:?}");
|
||||||
}
|
}
|
||||||
@@ -398,6 +413,8 @@ async fn run_live(
|
|||||||
let Some(command) = command else { break; };
|
let Some(command) = command else { break; };
|
||||||
match command {
|
match command {
|
||||||
RuntimeControlCommand::Pause => {
|
RuntimeControlCommand::Pause => {
|
||||||
|
live.landmark_roaming
|
||||||
|
.update_pause(|pause| pause.operator = true);
|
||||||
if let Err(error) = handle.control(SessionControl::Pause).await {
|
if let Err(error) = handle.control(SessionControl::Pause).await {
|
||||||
control_failure = Some(error.to_string());
|
control_failure = Some(error.to_string());
|
||||||
break;
|
break;
|
||||||
@@ -408,6 +425,8 @@ async fn run_live(
|
|||||||
control_failure = Some(error.to_string());
|
control_failure = Some(error.to_string());
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
live.landmark_roaming
|
||||||
|
.update_pause(|pause| pause.operator = false);
|
||||||
}
|
}
|
||||||
RuntimeControlCommand::ForceReconnect => {
|
RuntimeControlCommand::ForceReconnect => {
|
||||||
if let Err(error) = handle.control(SessionControl::ForceReconnect).await {
|
if let Err(error) = handle.control(SessionControl::ForceReconnect).await {
|
||||||
@@ -435,6 +454,7 @@ async fn run_live(
|
|||||||
.result_code("requested")?;
|
.result_code("requested")?;
|
||||||
let _ = live.observability.record(shutdown_event);
|
let _ = live.observability.record(shutdown_event);
|
||||||
control_target.mark_stopping();
|
control_target.mark_stopping();
|
||||||
|
live.landmark_roaming.shutdown().await;
|
||||||
let session_result = handle.shutdown().await;
|
let session_result = handle.shutdown().await;
|
||||||
let interaction_result = live.interaction.shutdown().await;
|
let interaction_result = live.interaction.shutdown().await;
|
||||||
let behavior_result = live.behavior.shutdown().await;
|
let behavior_result = live.behavior.shutdown().await;
|
||||||
@@ -468,6 +488,8 @@ struct LiveInteractions {
|
|||||||
policy: Arc<metacrate_grid_agent::PolicyGateway>,
|
policy: Arc<metacrate_grid_agent::PolicyGateway>,
|
||||||
audit: Arc<metacrate_grid_agent::MemoryPolicyAudit>,
|
audit: Arc<metacrate_grid_agent::MemoryPolicyAudit>,
|
||||||
observability: Arc<metacrate_grid_agent::Observability>,
|
observability: Arc<metacrate_grid_agent::Observability>,
|
||||||
|
_landmark_intake: metacrate_grid_agent::LibremetaverseLandmarkIntake,
|
||||||
|
landmark_roaming: metacrate_grid_agent::LandmarkRoamingHandle,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(feature = "live-grid")]
|
#[cfg(feature = "live-grid")]
|
||||||
@@ -478,11 +500,13 @@ fn start_live_interactions(
|
|||||||
) -> Result<LiveInteractions, Box<dyn Error>> {
|
) -> Result<LiveInteractions, Box<dyn Error>> {
|
||||||
use metacrate_grid_agent::{
|
use metacrate_grid_agent::{
|
||||||
AuthorizedBackendRouter, AuthorizedToolBackend, BehaviorBackend, BehaviorController,
|
AuthorizedBackendRouter, AuthorizedToolBackend, BehaviorBackend, BehaviorController,
|
||||||
ConversationStore, InteractionCoordinator, LibremetaverseScriptInventory, LlmClient,
|
ConversationStore, InteractionCoordinator, LandmarkLimits, LandmarkService,
|
||||||
LlmTransportLimits, MemoryPolicyAudit, Observability, ObservabilityLimits,
|
LandmarkToolBackend, LibremetaverseLandmarkGrid, LibremetaverseLandmarkIntake,
|
||||||
PerceptionBackend, PolicyAuditSink, PolicyGateway, PolicyLimits, PolicyLlmResponder,
|
LibremetaverseScriptInventory, LlmClient, LlmTransportLimits, MemoryPolicyAudit,
|
||||||
ScriptDeliveryBackend, ScriptDeliverySettings, ToolLoopLimits, UnifiedPolicyAudit,
|
Observability, ObservabilityLimits, PerceptionBackend, PolicyAuditSink, PolicyGateway,
|
||||||
behavior_policy_tools, perception_policy_tools, script_delivery_policy_tool,
|
PolicyLimits, PolicyLlmResponder, ScriptDeliveryBackend, ScriptDeliverySettings,
|
||||||
|
SystemRoamingRandom, ToolLoopLimits, UnifiedPolicyAudit, behavior_policy_tools,
|
||||||
|
landmark_policy_tools, perception_policy_tools, script_delivery_policy_tool,
|
||||||
};
|
};
|
||||||
|
|
||||||
let transport_limits = LlmTransportLimits {
|
let transport_limits = LlmTransportLimits {
|
||||||
@@ -534,6 +558,26 @@ fn start_live_interactions(
|
|||||||
tools.push(script_delivery_policy_tool(
|
tools.push(script_delivery_policy_tool(
|
||||||
ScriptDeliverySettings::default(),
|
ScriptDeliverySettings::default(),
|
||||||
)?);
|
)?);
|
||||||
|
let landmark_service = Arc::new(LandmarkService::new(
|
||||||
|
Arc::new(LibremetaverseLandmarkGrid::new(owner)),
|
||||||
|
Arc::new(SystemRoamingRandom::default()),
|
||||||
|
config.authorized_avatar_uuids.clone(),
|
||||||
|
LandmarkLimits::default(),
|
||||||
|
Some(config.storage_path.join("landmarks.json")),
|
||||||
|
)?);
|
||||||
|
let landmark_backend: Arc<dyn AuthorizedToolBackend> = Arc::new(
|
||||||
|
LandmarkToolBackend::new(Arc::clone(&landmark_service))
|
||||||
|
.with_behavior(behavior_ingress.clone()),
|
||||||
|
);
|
||||||
|
let landmark_intake = LibremetaverseLandmarkIntake::start(&landmark_service, owner)?;
|
||||||
|
let landmark_roaming = landmark_service.start_roaming_runner(
|
||||||
|
metacrate_grid_agent::RoamingPause {
|
||||||
|
degraded: true,
|
||||||
|
..metacrate_grid_agent::RoamingPause::default()
|
||||||
|
},
|
||||||
|
Some(behavior_ingress.clone()),
|
||||||
|
);
|
||||||
|
tools.extend(landmark_policy_tools(LandmarkLimits::default())?);
|
||||||
let routes = tools
|
let routes = tools
|
||||||
.iter()
|
.iter()
|
||||||
.map(|tool| tool.definition.name.as_str().to_owned())
|
.map(|tool| tool.definition.name.as_str().to_owned())
|
||||||
@@ -541,6 +585,8 @@ fn start_live_interactions(
|
|||||||
let backend: Arc<dyn AuthorizedToolBackend> =
|
let backend: Arc<dyn AuthorizedToolBackend> =
|
||||||
if name == metacrate_grid_agent::SCRIPT_DELIVERY_TOOL {
|
if name == metacrate_grid_agent::SCRIPT_DELIVERY_TOOL {
|
||||||
Arc::clone(&script_backend)
|
Arc::clone(&script_backend)
|
||||||
|
} else if name.starts_with("landmark_") {
|
||||||
|
Arc::clone(&landmark_backend)
|
||||||
} else if name.starts_with("behavior_") {
|
} else if name.starts_with("behavior_") {
|
||||||
Arc::new(BehaviorBackend::new(behavior_ingress.clone()))
|
Arc::new(BehaviorBackend::new(behavior_ingress.clone()))
|
||||||
} else {
|
} else {
|
||||||
@@ -601,6 +647,8 @@ fn start_live_interactions(
|
|||||||
policy: gateway,
|
policy: gateway,
|
||||||
audit,
|
audit,
|
||||||
observability,
|
observability,
|
||||||
|
_landmark_intake: landmark_intake,
|
||||||
|
landmark_roaming,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -48,10 +48,10 @@ fn package_has_only_reviewed_rust_dependencies_and_no_build_script() {
|
|||||||
#[test]
|
#[test]
|
||||||
fn runtime_source_has_no_subprocess_or_native_abi_escape_hatch() {
|
fn runtime_source_has_no_subprocess_or_native_abi_escape_hatch() {
|
||||||
let source = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("src");
|
let source = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("src");
|
||||||
let mut files = Vec::with_capacity(20);
|
let mut files = Vec::with_capacity(32);
|
||||||
collect_rust_files(&source, &mut files);
|
collect_rust_files(&source, &mut files);
|
||||||
assert!(
|
assert!(
|
||||||
files.len() <= 30,
|
files.len() <= 32,
|
||||||
"source-file count needs a reviewed bound update"
|
"source-file count needs a reviewed bound update"
|
||||||
);
|
);
|
||||||
for path in files {
|
for path in files {
|
||||||
|
|||||||
40
docs/grid-agent-landmarks.md
Normal file
40
docs/grid-agent-landmarks.md
Normal file
@@ -0,0 +1,40 @@
|
|||||||
|
# Landmark intake, teleport, and roaming
|
||||||
|
|
||||||
|
Landmark authority is private. Public chat and non-allow-listed IM can neither
|
||||||
|
accept offers nor see teleport/schedule tools. The policy layer seals the
|
||||||
|
authenticated avatar into every mutation; tool schemas contain a catalog
|
||||||
|
selector or bounded interval, never an avatar, region, coordinate, inventory
|
||||||
|
folder, or L$ field.
|
||||||
|
|
||||||
|
The live adapter subscribes through the original LibreMetaverse compatibility
|
||||||
|
events but lives entirely in `metacrate-grid-agent`. Authorized landmark and
|
||||||
|
folder offers are accepted into `MetaCrate Received Landmarks`, which is a
|
||||||
|
quarantine and recovery folder. Direct items are cataloged only after the
|
||||||
|
server returns authoritative item metadata. Folder descendants are fetched
|
||||||
|
with bounded breadth/depth and asset-fetch counts; a cycle, link, duplicate
|
||||||
|
asset, bad type, empty folder, stale response, or limit breach leaves the
|
||||||
|
quarantined inventory untouched and out of the catalog. Task offers and all
|
||||||
|
unauthorized or arbitrary inventory offers are declined.
|
||||||
|
|
||||||
|
The persisted catalog contains sender UUID, inventory UUID, asset UUID,
|
||||||
|
permission fingerprint, display name, receipt time, validation state, and last
|
||||||
|
outcome. Names are untrusted. Persistence is bounded, versioned, atomically
|
||||||
|
replaced, and mode `0600` on Unix. Teleport selection prefers stable IDs and
|
||||||
|
requires clarification for duplicate names. Immediately before the native
|
||||||
|
landmark teleport, the item is fetched again and its type, asset UUID, and
|
||||||
|
permissions must match. One semaphore, a timeout, cooldown, and lifecycle
|
||||||
|
cancellation prevent overlapping or stale teleports.
|
||||||
|
|
||||||
|
Roaming schedules retain the authorizing avatar UUID and safe minimum/maximum
|
||||||
|
intervals. Randomness is injectable. A due run chooses without immediate
|
||||||
|
repetition when possible and schedules its next deadline from the current time,
|
||||||
|
so downtime never creates catch-up bursts. Conversation, operator, degraded,
|
||||||
|
build, and viewport-capture pause reasons all suppress selection. Disabling a
|
||||||
|
schedule is persistent and cancellation stops in-flight native teleport work.
|
||||||
|
|
||||||
|
Focused verification:
|
||||||
|
|
||||||
|
```console
|
||||||
|
cargo test --locked -p metacrate-grid-agent --lib landmark_tests
|
||||||
|
cargo check --locked -p metacrate-grid-agent --all-targets --features live-grid
|
||||||
|
```
|
||||||
Reference in New Issue
Block a user