Implement authorized landmark roaming (#130)

crates/metacrate-grid-agent/src/landmark_tests.rs

@@ -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()
+    );
+}