diff --git a/crates/tracedecay-agent-hosts/src/native_integration/registry.rs b/crates/tracedecay-agent-hosts/src/native_integration/registry.rs index 0fb5a04cd7..2339ee3b5e 100644 --- a/crates/tracedecay-agent-hosts/src/native_integration/registry.rs +++ b/crates/tracedecay-agent-hosts/src/native_integration/registry.rs @@ -29,9 +29,8 @@ use tracedecay_contracts::git::{ use tracedecay_contracts::{ AuthorizedScopeSet, CancellationSignal, NativeIntegrationContractError, NativeIntegrationPort, NativeIntegrationPortError, NativeIntegrationRecoveryRequestV1, NativeIntegrationService, - NativeIntegrationStackResolutionOutcomeV1, NativeIntegrationStackResolutionPort, - NativeIntegrationStackResolutionRequestV1, NativeIntegrationStackSnapshotService, - ResolvedScope, + NativeIntegrationStackResolutionOutcomeV1, NativeIntegrationStackResolutionRequestV1, + NativeIntegrationStackSnapshotService, ResolvedScope, }; use tracedecay_domain::{ ManifestDigest, ProjectId, RepositoryId, ScopeSetId, ScopeSetRevision, UtcMicros, @@ -52,7 +51,7 @@ const MAX_PENDING_WORKTREE_CLEANUPS: u32 = 4_096; /// The one exact composition served to invocation routing. pub type DaemonProjectNativeIntegrationCoordinator = NativeIntegrationTransactionCoordinator< SharedDaemonNativeIntegrationStore, - SharedProjectNativeIntegrationTopology, + ExactPairNativeIntegrationTopology, GixNativeIntegrationAdapter, DaemonNativeIntegrationAuthorization, >; @@ -80,23 +79,6 @@ fn worktree_recovery_error(error: WorktreeContractError) -> NativeIntegrationPor } } -/// Shares one enrolled topology resolver between the transaction coordinator -/// and the stack-snapshot service without a second repository handle. -#[derive(Clone)] -pub struct SharedProjectNativeIntegrationTopology { - inner: Arc, -} - -impl NativeIntegrationStackResolutionPort for SharedProjectNativeIntegrationTopology { - fn resolve( - &self, - request: &NativeIntegrationStackResolutionRequestV1, - cancellation: &CancellationSignal, - ) -> Result { - self.inner.resolve(request, cancellation) - } -} - /// Retains the one `DaemonNativeIntegrationStore` actor for each daemon-owned /// project database. Dropping the registry closes every actor when the daemon /// store administration shuts down. @@ -175,7 +157,7 @@ pub struct DaemonNativeIntegrationOwner { pub project_id: ProjectId, pub repository_id: RepositoryId, service: Arc, - snapshots: Arc>, + snapshots: Arc>>, worktrees: Option>, store: SharedDaemonNativeIntegrationStore, scope_sets: Option, @@ -508,25 +490,23 @@ impl DaemonNativeIntegrationServiceRegistry { let owner_repository_id = repository_id.clone(); let (owner_project_id, owner_repository_id, service, snapshots, worktrees) = tokio::task::spawn_blocking(move || { - let topology = SharedProjectNativeIntegrationTopology { - inner: Arc::new(match (topology_shard, topology_runtime) { - (Some(expected_shard), Some(runtime)) => { - ExactPairNativeIntegrationTopology::open_with_graph_runtime_provider( - owner_project_id.clone(), - owner_repository_id.clone(), - &native_root, - expected_shard, - runtime, - )? - } - (None, None) => ExactPairNativeIntegrationTopology::open( + let topology = Arc::new(match (topology_shard, topology_runtime) { + (Some(expected_shard), Some(runtime)) => { + ExactPairNativeIntegrationTopology::open_with_graph_runtime_provider( owner_project_id.clone(), owner_repository_id.clone(), &native_root, - )?, - _ => return Err(NativeIntegrationPortError::Unavailable), - }), - }; + expected_shard, + runtime, + )? + } + (None, None) => ExactPairNativeIntegrationTopology::open( + owner_project_id.clone(), + owner_repository_id.clone(), + &native_root, + )?, + _ => return Err(NativeIntegrationPortError::Unavailable), + }); let native = GixNativeIntegrationAdapter::open( owner_project_id.clone(), owner_repository_id.clone(), @@ -536,7 +516,7 @@ impl DaemonNativeIntegrationServiceRegistry { .map_err(|_| NativeIntegrationPortError::Unavailable)?; let coordinator = NativeIntegrationTransactionCoordinator::new( Arc::new(recovery_store.clone()), - Arc::new(topology.clone()), + Arc::clone(&topology), Arc::new(native), Arc::new(authorization), ); @@ -573,7 +553,9 @@ impl DaemonNativeIntegrationServiceRegistry { owner_project_id, owner_repository_id, Arc::new(NativeIntegrationService::new(coordinator)), - Arc::new(NativeIntegrationStackSnapshotService::new(topology)), + Arc::new(NativeIntegrationStackSnapshotService::new(Arc::clone( + &topology, + ))), worktrees, )) }) @@ -679,17 +661,25 @@ impl DaemonNativeIntegrationServiceRegistry { #[cfg(test)] mod tests { + use std::collections::BTreeSet; use std::path::Path; use std::process::Command; use std::sync::Arc; use tracedecay_contracts::{ - NativeIntegrationCancelDispositionV1, NativeIntegrationCancelRequestV1, - NativeIntegrationStatusRequestV1, + AuthorizedScopeSetAuthority, CancellationContext, CancellationSignal, CapabilityGrantId, + CapabilityGrantSnapshot, Deadline, DisclosureClass, NativeIntegrationCancelDispositionV1, + NativeIntegrationCancelRequestV1, NativeIntegrationSelectionBindingV1, + NativeIntegrationStackResolutionOutcomeV1, NativeIntegrationStackResolutionRequestV1, + NativeIntegrationStatusRequestV1, RequestContext, RequestId, ResolvedScope, + native_integration_surface_operation, }; use tracedecay_domain::{ - ManifestDigest, NativeIntegrationTransactionId, ProjectId, RepositoryId, UtcMicros, + ActorId, ManifestDigest, NativeIntegrationTransactionId, ProjectId, RefId, RepositoryId, + ScopeSetId, ScopeSetRevision, UtcMicros, WorktreeId, WorktreeInventoryEpoch, + WorktreeInventorySnapshotId, }; + use tracedecay_tool_catalog::{CapabilityId, UseCaseId}; use super::DaemonNativeIntegrationServiceRegistry; use tracedecay_global_db::tests::harness::HostAdmissionTestRuntimeV1; @@ -716,6 +706,262 @@ mod tests { ManifestDigest::new(format!("sha256:{}", "5".repeat(64))).expect("policy digest") } + fn digest(byte: char) -> ManifestDigest { + ManifestDigest::new(format!("sha256:{}", byte.to_string().repeat(64))).expect("digest") + } + + fn git(root: &Path, arguments: &[&str]) { + let status = Command::new(try_git_program().expect("resolve the git program")) + .args(arguments) + .current_dir(root) + .status() + .expect("run git fixture command"); + assert!(status.success(), "git {arguments:?} failed"); + } + + fn prepare_independent_pair(root: &Path) { + init_repository(root); + git(root, &["checkout", "-b", "destination"]); + git(root, &["checkout", "main"]); + git(root, &["checkout", "-b", "source"]); + std::fs::write(root.join("source.txt"), "source\n").expect("write source"); + git(root, &["add", "source.txt"]); + git(root, &["commit", "-m", "source"]); + git(root, &["checkout", "main"]); + } + + fn exact_pair_scopes( + project: &ProjectId, + repository: &RepositoryId, + ) -> (ResolvedScope, ResolvedScope) { + let source = ResolvedScope::new( + project.clone(), + repository.clone(), + WorktreeId::new("worktree.native.snapshot.source").expect("source worktree id"), + Some(RefId::new("refs/heads/source").expect("source ref")), + ) + .expect("source scope"); + let destination = ResolvedScope::new( + project.clone(), + repository.clone(), + WorktreeId::new("worktree.native.snapshot.destination") + .expect("destination worktree id"), + Some(RefId::new("refs/heads/destination").expect("destination ref")), + ) + .expect("destination scope"); + (source, destination) + } + + fn stack_snapshot_request( + project: ProjectId, + repository: RepositoryId, + source: ResolvedScope, + destination: ResolvedScope, + ) -> NativeIntegrationStackResolutionRequestV1 { + let (capability, use_case) = { + let operation = native_integration_surface_operation( + tracedecay_contracts::NATIVE_INTEGRATION_STACK_SNAPSHOT_OPERATION, + ) + .expect("canonical operation") + .expect("declared operation"); + ( + operation.capability_id().clone(), + operation.use_case_id().clone(), + ) + }; + let grant = CapabilityGrantSnapshot::new( + CapabilityGrantId::new("grant.native.snapshot").expect("grant id"), + 1, + digest('a'), + ActorId::new("actor.native.issuer").expect("issuer"), + UtcMicros(1), + UtcMicros(10_000), + destination.clone(), + BTreeSet::from([capability.clone()]), + BTreeSet::from([use_case.clone()]), + DisclosureClass::Sensitive, + ) + .expect("grant"); + let context = RequestContext::new( + ActorId::new("actor.native.requester").expect("requester"), + destination.clone(), + grant, + RequestId::new("request.native.snapshot").expect("request id"), + Deadline::new(UtcMicros(10_000)).expect("deadline"), + CancellationContext::active("cancel.native.snapshot").expect("cancellation"), + ) + .expect("request context"); + let source_grant = CapabilityGrantSnapshot::new( + CapabilityGrantId::new("grant.native.snapshot.source").expect("grant id"), + 1, + digest('a'), + ActorId::new("actor.native.issuer").expect("issuer"), + UtcMicros(1), + UtcMicros(10_000), + source.clone(), + BTreeSet::from([capability.clone()]), + BTreeSet::from([use_case.clone()]), + DisclosureClass::Sensitive, + ) + .expect("source grant"); + let authorized_scope_set = AuthorizedScopeSetAuthority::authorize( + ScopeSetId::new("scope-set.native.snapshot").expect("scope set id"), + ScopeSetRevision::new(1).expect("scope set revision"), + vec![ + context, + RequestContext::new( + ActorId::new("actor.native.requester").expect("requester"), + source.clone(), + source_grant, + RequestId::new("request.native.snapshot.source").expect("request id"), + Deadline::new(UtcMicros(10_000)).expect("deadline"), + CancellationContext::active("cancel.native.snapshot.source") + .expect("cancellation"), + ) + .expect("source context"), + ], + &capability, + &use_case, + UtcMicros(100), + ) + .expect("authorized scope set"); + NativeIntegrationStackResolutionRequestV1 { + source, + destination, + authorized_scope_set, + inventory_snapshot_id: WorktreeInventorySnapshotId::new("inventory.native.snapshot") + .expect("inventory snapshot"), + inventory_epoch: WorktreeInventoryEpoch::new(1).expect("inventory epoch"), + selection: NativeIntegrationSelectionBindingV1::IndependentBranch { + proposal_digest: digest('c'), + }, + grant_digest: digest('a'), + policy_digest: digest('d'), + observed_at: UtcMicros(100), + } + } + + #[tokio::test(flavor = "multi_thread")] + async fn ensure_stack_snapshot_freezes_topology_and_honours_guardrails() { + let directory = tempfile::tempdir().expect("temporary project directory"); + let repository_root = directory.path().join("repo"); + std::fs::create_dir_all(&repository_root).expect("repository root"); + prepare_independent_pair(&repository_root); + let registry = DaemonNativeIntegrationServiceRegistry::default(); + let project_id = ProjectId::new("project.native.snapshot").expect("project id"); + let repository_id = RepositoryId::new("repository.native.snapshot").expect("repository id"); + let runtime = HostAdmissionTestRuntimeV1::project( + directory.path().join("profile"), + &repository_root, + project_id.clone(), + ) + .await + .expect("canonical project test runtime"); + let database = runtime + .registered_database_lease(HostAdmissionScope::Project) + .expect("registered project database"); + let owner = registry + .ensure( + database, + repository_root, + project_id.clone(), + repository_id.clone(), + policy_digest(), + UtcMicros(100), + ) + .await + .expect("mount native integration owner"); + + let (source, destination) = exact_pair_scopes(&project_id, &repository_id); + let request = stack_snapshot_request( + project_id.clone(), + repository_id.clone(), + source, + destination, + ); + let topology_request = request.clone(); + let signal = CancellationSignal::active("cancel.native.snapshot.resolve").expect("signal"); + let snapshot_owner = owner.clone(); + let outcome = tokio::task::spawn_blocking(move || { + snapshot_owner.stack_snapshot(topology_request, &signal) + }) + .await + .expect("stack snapshot join") + .expect("stack snapshot result"); + let NativeIntegrationStackResolutionOutcomeV1::Complete(frozen) = outcome else { + panic!("ensure must expose stack snapshot through the retained topology: {outcome:?}"); + }; + let selection = frozen.as_ref(); + assert_eq!(selection.project_id().expect("project"), &project_id); + assert_eq!( + selection.repository_id().expect("repository"), + &repository_id + ); + match selection { + tracedecay_domain::NativeIntegrationSelectionV1::IndependentBranch(branch) => { + assert!(branch.source_worktree_id.is_none()); + assert!(branch.destination_worktree_id.is_none()); + } + other => { + panic!("independent pair fixture must freeze an independent branch: {other:?}") + } + } + + let second_owner = owner.clone(); + let second_request = request.clone(); + let second_signal = + CancellationSignal::active("cancel.native.snapshot.second").expect("signal"); + let second_outcome = tokio::task::spawn_blocking(move || { + second_owner.stack_snapshot(second_request, &second_signal) + }) + .await + .expect("second stack snapshot join") + .expect("second stack snapshot result"); + let NativeIntegrationStackResolutionOutcomeV1::Complete(second_frozen) = second_outcome + else { + panic!("retained topology must resolve consistently: {second_outcome:?}"); + }; + assert_eq!(second_frozen.as_ref(), frozen.as_ref()); + + let foreign_project = ProjectId::new("project.native.snapshot.foreign").expect("foreign"); + let (foreign_source, foreign_destination) = + exact_pair_scopes(&foreign_project, &repository_id); + let foreign_request = stack_snapshot_request( + foreign_project, + repository_id.clone(), + foreign_source, + foreign_destination, + ); + let denied_owner = owner.clone(); + let denied_signal = + CancellationSignal::active("cancel.native.snapshot.denied").expect("signal"); + let denied = tokio::task::spawn_blocking(move || { + denied_owner.stack_snapshot(foreign_request, &denied_signal) + }) + .await + .expect("denied join") + .expect("denied result"); + assert_eq!( + denied, + NativeIntegrationStackResolutionOutcomeV1::Denied, + "foreign project identity must not resolve against the enrolled topology" + ); + + let cancelled_signal = + CancellationSignal::active("cancel.native.snapshot.already").expect("signal"); + cancelled_signal.cancel(UtcMicros(99)); + let cancelled = owner + .stack_snapshot(request, &cancelled_signal) + .expect("cancelled stack snapshot"); + assert_eq!( + cancelled, + NativeIntegrationStackResolutionOutcomeV1::Unavailable, + "cancellation must fail closed before topology resolution" + ); + + registry.shutdown().await.expect("shutdown"); + } + #[tokio::test(flavor = "multi_thread")] async fn mounts_one_owner_and_answers_typed_states_for_unknown_transactions() { let directory = tempfile::tempdir().expect("temporary project directory"); diff --git a/crates/tracedecay-contracts/src/git/native_integration.rs b/crates/tracedecay-contracts/src/git/native_integration.rs index fd76f0051f..769ed94eb5 100644 --- a/crates/tracedecay-contracts/src/git/native_integration.rs +++ b/crates/tracedecay-contracts/src/git/native_integration.rs @@ -4,6 +4,8 @@ //! Filesystem paths, free-form object IDs, Git arguments, commit messages, //! remotes, and provider mutations are intentionally unrepresentable. +use std::sync::Arc; + use schemars::JsonSchema; use serde::{Deserialize, Serialize}; use thiserror::Error; @@ -190,6 +192,18 @@ pub trait NativeIntegrationStackResolutionPort: Send + Sync { ) -> Result; } +impl NativeIntegrationStackResolutionPort + for Arc +{ + fn resolve( + &self, + request: &NativeIntegrationStackResolutionRequestV1, + cancellation: &CancellationSignal, + ) -> Result { + self.as_ref().resolve(request, cancellation) + } +} + /// Exact semantic evidence revisions joined to native conflict evidence. #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] #[serde(deny_unknown_fields)]