feat(core,call): implement ADR-050 dynamic resource ownership — ownership store, resource_id_path, check signature, dispatch wiring

Implements the 4-task DAG for runtime-spawned resource ownership so
AccessControl::check can answer "does this identity own this specific
container/TTY/process" instead of relying only on static
Identity.resources grants.

1. core/ownership-store-trait: OwnershipProvider (sync read) +
   OwnershipStore (async write) traits + InMemoryOwnershipStore +
   OwnershipError in alknet-core. Fourth instance of the repo/adapter
   pattern (ADR-033), mirroring CredentialStore/store.rs.

2. call/registry/operation-spec-resource-id-path: resource_id_path
   field on OperationSpec — JSON pointer into input for resource ID
   extraction. Single field addition, all ~40 construction sites across
   alknet-call + alknet-http updated to pass None (no semantic change).

3. call/registry/access-control-ownership-check: check() signature
   gains resource_id + Option<&dyn OwnershipProvider>. 3-case decision
   tree: ownership Some + resource_id Some -> owns(); ownership Some +
   resource_id None -> owns_any() (list scope-gate); ownership None ->
   static Identity.resources fallback (backward compat). 7 call sites
   updated to (None, None) — including a 7th in alknet-http/gateway_routes
   not listed in the original spec.

4. call/registry/dispatch-resource-id-extraction: wire the dispatch
   path. OperationContext gains ownership field; Dispatcher/CallAdapter
   gain with_ownership_provider builders; invoke/invoke_streaming/
   invoke_with_policy extract resource_id via spec.resource_id_path
   and thread context.ownership to check(). extract_json_pointer helper
   handles $.field syntax (graceful None on missing/non-string). 20
   OperationContext literals updated across both crates.

Backward compatibility is load-bearing throughout: ownership=None
falls back to the existing static resource-check path. Deployments
without runtime-spawned resources wire nothing and behave identically.

821 tests pass workspace-wide (was ~770); clippy clean; fmt clean.
Task specs marked done.

crates/alknet-call/src/client/call_client.rs

@@ -615,6 +615,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             AccessControl::default(),
+            None,
         )
     }
 

crates/alknet-call/src/client/from_call.rs

@@ -253,6 +253,7 @@ fn rebuild_spec_for(
         output_schema,
         error_schemas,
         access_control,
+        None,
     ))
 }
 
@@ -582,6 +583,7 @@ mod tests {
             json!({}),
             vec![],
             AccessControl::default(),
+            None,
         );
         let handler = make_forwarding_handler(
             Arc::new(CallConnection::new(stub_connection())),
@@ -638,6 +640,7 @@ mod tests {
             abort_policy: AbortPolicy::default(),
             deadline: Some(Instant::now() + Duration::from_secs(30)),
             internal: false,
+            ownership: None,
         }
     }
 
@@ -1090,6 +1093,7 @@ mod tests {
                 json!({}),
                 vec![],
                 AccessControl::default(),
+                None,
             ),
             HandlerKind::Stream(handler),
             OperationProvenance::FromCall,

crates/alknet-call/src/client/from_jsonschema.rs

@@ -107,6 +107,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             AccessControl::default(),
+            None,
         )
     }
 
@@ -124,6 +125,7 @@ mod tests {
             abort_policy: AbortPolicy::default(),
             deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
             internal: true,
+            ownership: None,
         }
     }
 

crates/alknet-call/src/protocol/adapter.rs

@@ -14,6 +14,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use alknet_core::auth::{AuthContext, IdentityProvider};
+use alknet_core::ownership::OwnershipProvider;
 use alknet_core::types::{Connection, HandlerError, ProtocolHandler};
 use async_trait::async_trait;
 
@@ -63,6 +64,11 @@ impl CallAdapter {
         self
     }
 
+    pub fn with_ownership_provider(mut self, provider: Arc<dyn OwnershipProvider>) -> Self {
+        self.dispatcher = self.dispatcher.with_ownership_provider(provider);
+        self
+    }
+
     pub fn registry(&self) -> &Arc<OperationRegistry> {
         &self.dispatcher.registry
     }
@@ -224,6 +230,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             acl,
+            None,
         )
     }
 
@@ -237,6 +244,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             AccessControl::default(),
+            None,
         )
     }
 
@@ -257,6 +265,7 @@ mod tests {
                     serde_json::json!({}),
                     vec![],
                     acl,
+                    None,
                 ),
                 HandlerKind::Once(handler),
                 OperationProvenance::Local,
@@ -548,6 +557,7 @@ mod tests {
                 serde_json::json!({}),
                 vec![],
                 AccessControl::default(),
+                None,
             ),
             HandlerKind::Once(echo_handler()),
             OperationProvenance::FromCall,
@@ -583,6 +593,7 @@ mod tests {
             abort_policy: AbortPolicy::default(),
             deadline: context.deadline,
             internal: false,
+            ownership: None,
         };
         let response = context
             .env
@@ -615,6 +626,7 @@ mod tests {
                 serde_json::json!({}),
                 vec![],
                 AccessControl::default(),
+                None,
             ),
             HandlerKind::Once(echo_handler()),
             OperationProvenance::FromCall,
@@ -641,6 +653,7 @@ mod tests {
             abort_policy: AbortPolicy::default(),
             deadline: context.deadline,
             internal: false,
+            ownership: None,
         };
         let response = context
             .env

crates/alknet-call/src/protocol/connection.rs

@@ -326,6 +326,7 @@ impl OperationEnv for OverlayOperationEnv {
         let composition_authority;
         let scoped_env;
         let access_control;
+        let resource_id_path;
         {
             let overlay = self.overlay.read();
             let Some(registration) = overlay.get(&name) else {
@@ -338,6 +339,7 @@ impl OperationEnv for OverlayOperationEnv {
                 .clone()
                 .unwrap_or_else(ScopedPeerEnv::empty);
             access_control = registration.spec.access_control.clone();
+            resource_id_path = registration.spec.resource_id_path.clone();
         }
 
         let caller_identity = if parent.internal {
@@ -348,7 +350,14 @@ impl OperationEnv for OverlayOperationEnv {
         } else {
             parent.identity.clone()
         };
-        if let AccessResult::Forbidden(message) = access_control.check(caller_identity.as_ref()) {
+        let resource_id = resource_id_path
+            .as_ref()
+            .and_then(|path| crate::registry::registration::extract_json_pointer(&input, path));
+        if let AccessResult::Forbidden(message) = access_control.check(
+            caller_identity.as_ref(),
+            resource_id.as_deref(),
+            parent.ownership.as_deref(),
+        ) {
             return ResponseEnvelope::forbidden(parent.request_id.clone(), message);
         }
 
@@ -368,6 +377,7 @@ impl OperationEnv for OverlayOperationEnv {
             scoped_env,
             env: parent.env.clone(),
             internal: true,
+            ownership: parent.ownership.clone(),
         };
 
         match handler {
@@ -487,6 +497,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             AccessControl::default(),
+            None,
         )
     }
 
@@ -525,6 +536,7 @@ mod tests {
             abort_policy: AbortPolicy::default(),
             deadline: Some(Instant::now() + Duration::from_secs(30)),
             internal: true,
+            ownership: None,
         }
     }
 
@@ -1002,6 +1014,7 @@ mod tests {
                 serde_json::json!({}),
                 vec![],
                 AccessControl::default(),
+                None,
             ),
             HandlerKind::Stream(streaming_handler),
             OperationProvenance::FromCall,

crates/alknet-call/src/protocol/dispatch.rs

@@ -16,6 +16,7 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use alknet_core::auth::{AuthToken, Identity, IdentityProvider};
+use alknet_core::ownership::OwnershipProvider;
 use alknet_core::types::StreamError;
 use futures::stream::StreamExt;
 use serde_json::Value;
@@ -70,6 +71,7 @@ pub struct Dispatcher {
     pub registry: Arc<OperationRegistry>,
     pub identity_provider: Arc<dyn IdentityProvider>,
     pub session_source: Option<Arc<dyn SessionOverlaySource + Send + Sync>>,
+    pub ownership_provider: Option<Arc<dyn OwnershipProvider>>,
     pub default_timeout: Duration,
 }
 
@@ -82,6 +84,7 @@ impl Dispatcher {
             registry,
             identity_provider,
             session_source: None,
+            ownership_provider: None,
             default_timeout: DEFAULT_TIMEOUT,
         }
     }
@@ -99,6 +102,11 @@ impl Dispatcher {
         self
     }
 
+    pub fn with_ownership_provider(mut self, provider: Arc<dyn OwnershipProvider>) -> Self {
+        self.ownership_provider = Some(provider);
+        self
+    }
+
     fn strip_leading_slash(operation_id: &str) -> &str {
         operation_id.strip_prefix('/').unwrap_or(operation_id)
     }
@@ -182,6 +190,7 @@ impl Dispatcher {
             env: stub_env,
             abort_policy: AbortPolicy::default(),
             internal: false,
+            ownership: self.ownership_provider.clone(),
         };
         context.env = self.compose_root_env(connection, &context);
         context
@@ -432,6 +441,7 @@ impl Clone for Dispatcher {
             registry: Arc::clone(&self.registry),
             identity_provider: Arc::clone(&self.identity_provider),
             session_source: self.session_source.clone(),
+            ownership_provider: self.ownership_provider.clone(),
             default_timeout: self.default_timeout,
         }
     }
@@ -524,6 +534,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             acl,
+            None,
         )
     }
 
@@ -539,6 +550,7 @@ mod tests {
                     serde_json::json!({}),
                     vec![],
                     acl,
+                    None,
                 ),
                 HandlerKind::Once(make_handler(|input, context| async move {
                     ResponseEnvelope::ok(context.request_id, input)
@@ -1001,6 +1013,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             acl,
+            None,
         )
     }
 

crates/alknet-call/src/registry/context.rs

@@ -3,6 +3,7 @@ use std::sync::Arc;
 use std::time::Instant;
 
 use alknet_core::auth::Identity;
+use alknet_core::ownership::OwnershipProvider;
 use alknet_core::types::Capabilities;
 use serde_json::Value;
 
@@ -30,6 +31,10 @@ pub struct OperationContext {
     pub abort_policy: AbortPolicy,
     pub deadline: Option<Instant>,
     pub internal: bool,
+    /// `None` when no ownership provider is wired (backward compat —
+    /// `check` falls back to static `Identity.resources` path). Wired by
+    /// the assembly layer via `CallAdapter`/`Dispatcher` (ADR-050).
+    pub ownership: Option<Arc<dyn OwnershipProvider>>,
 }
 
 impl OperationContext {

crates/alknet-call/src/registry/discovery.rs

@@ -38,6 +38,7 @@ pub fn services_list_spec() -> OperationSpec {
         }),
         vec![],
         AccessControl::default(),
+        None,
     )
 }
 
@@ -54,6 +55,7 @@ pub fn services_schema_spec() -> OperationSpec {
         operation_spec_schema(),
         vec![],
         AccessControl::default(),
+        None,
     )
 }
 
@@ -93,6 +95,7 @@ pub fn services_list_peers_spec() -> OperationSpec {
         }),
         vec![],
         AccessControl::default(),
+        None,
     )
 }
 
@@ -221,7 +224,11 @@ pub fn services_list_handler(registry: Arc<OperationRegistry>) -> Handler {
             let ops: Vec<Value> = registry
                 .list_operations()
                 .into_iter()
-                .filter(|spec| spec.access_control.check(calling_identity).is_allowed())
+                .filter(|spec| {
+                    spec.access_control
+                        .check(calling_identity, None, None)
+                        .is_allowed()
+                })
                 .map(|s| {
                     json!({
                         "name": s.name,
@@ -244,7 +251,11 @@ pub fn services_list_peers_handler(registry: Arc<OperationRegistry>) -> Handler
             let local_ops: Vec<Value> = registry
                 .list_operations()
                 .into_iter()
-                .filter(|spec| spec.access_control.check(calling_identity).is_allowed())
+                .filter(|spec| {
+                    spec.access_control
+                        .check(calling_identity, None, None)
+                        .is_allowed()
+                })
                 .map(|s| {
                     json!({
                         "name": s.name,
@@ -265,9 +276,11 @@ pub fn services_list_peers_handler(registry: Arc<OperationRegistry>) -> Handler
                     .filter(|name| {
                         let spec = registry.registration(name);
                         match spec {
-                            Some(reg) => {
+                            Some(reg) => reg
-                                reg.spec.access_control.check(calling_identity).is_allowed()
+                                .spec
-                            }
+                                .access_control
+                                .check(calling_identity, None, None)
+                                .is_allowed(),
                             None => true,
                         }
                     })
@@ -341,6 +354,7 @@ mod tests {
             json!({}),
             vec![],
             AccessControl::default(),
+            None,
         )
     }
 
@@ -353,6 +367,7 @@ mod tests {
             json!({}),
             vec![],
             AccessControl::default(),
+            None,
         )
     }
 
@@ -403,6 +418,7 @@ mod tests {
             abort_policy: crate::registry::context::AbortPolicy::default(),
             deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
             internal: false,
+            ownership: None,
         }
     }
 
@@ -423,6 +439,7 @@ mod tests {
             abort_policy: crate::registry::context::AbortPolicy::default(),
             deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
             internal: false,
+            ownership: None,
         }
     }
 
@@ -443,6 +460,7 @@ mod tests {
             json!({}),
             vec![],
             acl,
+            None,
         )
     }
 
@@ -530,6 +548,7 @@ mod tests {
                     json!({}),
                     vec![],
                     AccessControl::default(),
+                    None,
                 ),
                 HandlerKind::Stream(echo_streaming_handler()),
                 OperationProvenance::Local,
@@ -553,6 +572,7 @@ mod tests {
                         http_status: None,
                     }],
                     AccessControl::default(),
+                    None,
                 ),
                 HandlerKind::Once(echo_handler()),
                 OperationProvenance::Local,
@@ -759,6 +779,7 @@ mod tests {
                 required_scopes: vec!["fs:read".to_string()],
                 ..Default::default()
             },
+            None,
         );
         let json_val = spec_to_json(&spec);
         let error_schemas = json_val
@@ -868,6 +889,7 @@ mod tests {
             abort_policy: crate::registry::context::AbortPolicy::default(),
             deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
             internal: false,
+            ownership: None,
         };
         let response = handler(serde_json::json!({}), ctx).await;
         let output = response.result.expect("ok response");
@@ -951,6 +973,7 @@ mod tests {
             abort_policy: crate::registry::context::AbortPolicy::default(),
             deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
             internal: false,
+            ownership: None,
         };
         let response = handler(serde_json::json!({}), ctx).await;
         let output = response.result.expect("ok response");

crates/alknet-call/src/registry/env.rs

@@ -139,6 +139,7 @@ impl OperationEnv for LocalOperationEnv {
                 .unwrap_or_else(ScopedPeerEnv::empty),
             env: parent.env.clone(),
             internal: true,
+            ownership: parent.ownership.clone(),
         };
 
         self.registry.invoke(&name, input, context).await
@@ -397,6 +398,7 @@ mod tests {
             abort_policy: AbortPolicy::default(),
             deadline: Some(Instant::now() + Duration::from_secs(30)),
             internal: false,
+            ownership: None,
         }
     }
 
@@ -418,6 +420,7 @@ mod tests {
                     serde_json::json!({}),
                     vec![],
                     AccessControl::default(),
+                    None,
                 ),
                 HandlerKind::Once(handler),
                 OperationProvenance::Local,

crates/alknet-call/src/registry/registration.rs

@@ -139,7 +139,17 @@ impl OperationRegistry {
             context.identity.clone()
         };
 
-        if let AccessResult::Forbidden(message) = acl.check(identity.as_ref()) {
+        let resource_id = registration
+            .spec
+            .resource_id_path
+            .as_ref()
+            .and_then(|path| extract_json_pointer(&input, path));
+
+        if let AccessResult::Forbidden(message) = acl.check(
+            identity.as_ref(),
+            resource_id.as_deref(),
+            context.ownership.as_deref(),
+        ) {
             return ResponseEnvelope::forbidden(request_id, message);
         }
 
@@ -191,7 +201,17 @@ impl OperationRegistry {
             context.identity.clone()
         };
 
-        if let AccessResult::Forbidden(message) = acl.check(identity.as_ref()) {
+        let resource_id = registration
+            .spec
+            .resource_id_path
+            .as_ref()
+            .and_then(|path| extract_json_pointer(&input, path));
+
+        if let AccessResult::Forbidden(message) = acl.check(
+            identity.as_ref(),
+            resource_id.as_deref(),
+            context.ownership.as_deref(),
+        ) {
             return Box::pin(stream::once(async move {
                 ResponseEnvelope::forbidden(request_id, message)
             }));
@@ -385,6 +405,38 @@ where
     Arc::new(move |input, context| Box::pin(f(input, context)))
 }
 
+/// Extract a string value from `input` at a JSON-pointer-ish path described
+/// by `$.field` or `$.field/sub` (a leading `$` followed by a slash-separated
+/// pointer, the same shape `serde_json::Value::pointer` expects after the
+/// leading `/` is restored). Also tolerates the dotted form `$.a.b` for
+/// shallow nested lookups (translated to `/a/b`).
+///
+/// Returns `None` if the path is malformed, the field is missing, or the
+/// value at the path is not a string. Graceful — never panics (ADR-050 §2a):
+/// a missing field simply yields `resource_id: None`, and the caller's
+/// `AccessControl::check` decides whether to deny based on whether the spec
+/// requires a specific resource ID.
+pub(crate) fn extract_json_pointer(input: &Value, path: &str) -> Option<String> {
+    let trimmed = path.strip_prefix('$')?;
+    if trimmed.is_empty() {
+        return None;
+    }
+    let pointer = if let Some(rest) = trimmed.strip_prefix('/') {
+        format!("/{}", rest)
+    } else {
+        let dotted = trimmed.replace('.', "/");
+        if let Some(rest) = dotted.strip_prefix('/') {
+            format!("/{}", rest)
+        } else {
+            format!("/{}", dotted)
+        }
+    };
+    input
+        .pointer(pointer.as_str())
+        .and_then(|v| v.as_str())
+        .map(|s| s.to_string())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -393,6 +445,7 @@ mod tests {
     use crate::registry::env::OperationEnv;
     use crate::registry::spec::{AccessControl, OperationType};
     use alknet_core::auth::Identity;
+    use alknet_core::ownership::OwnershipProvider;
     use std::collections::HashMap;
     use std::time::Duration;
 
@@ -436,6 +489,30 @@ mod tests {
             abort_policy: AbortPolicy::default(),
             deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
             internal,
+            ownership: None,
+        }
+    }
+
+    fn root_context_with_ownership(
+        request_id: &str,
+        identity: Option<Identity>,
+        ownership: Option<Arc<dyn OwnershipProvider>>,
+        scoped_env: ScopedPeerEnv,
+    ) -> OperationContext {
+        OperationContext {
+            request_id: request_id.to_string(),
+            parent_request_id: None,
+            identity,
+            handler_identity: None,
+            forwarded_for: None,
+            capabilities: Capabilities::new(),
+            metadata: HashMap::new(),
+            scoped_env,
+            env: Arc::new(NoopEnv),
+            abort_policy: AbortPolicy::default(),
+            deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
+            internal: false,
+            ownership,
         }
     }
 
@@ -460,6 +537,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             acl,
+            None,
         )
     }
 
@@ -472,6 +550,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             acl,
+            None,
         )
     }
 
@@ -948,6 +1027,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             AccessControl::default(),
+            None,
         )
     }
 
@@ -1234,6 +1314,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             acl,
+            None,
         )
     }
 
@@ -1246,6 +1327,308 @@ mod tests {
             serde_json::json!({}),
             vec![],
             acl,
+            None,
         )
     }
+
+    struct MockOwnership {
+        owned: Vec<(String, String)>,
+    }
+
+    impl OwnershipProvider for MockOwnership {
+        fn owns(
+            &self,
+            _identity: &Identity,
+            resource_type: &str,
+            resource_id: &str,
+            _action: &str,
+        ) -> bool {
+            self.owned
+                .iter()
+                .any(|(rt, rid)| rt == resource_type && rid == resource_id)
+        }
+
+        fn owned_resources(&self, _identity: &Identity, resource_type: &str) -> Vec<String> {
+            self.owned
+                .iter()
+                .filter(|(rt, _)| rt == resource_type)
+                .map(|(_, rid)| rid.clone())
+                .collect()
+        }
+
+        fn owns_any(&self, _identity: &Identity, resource_type: &str) -> bool {
+            self.owned.iter().any(|(rt, _)| rt == resource_type)
+        }
+    }
+
+    fn empty_identity(id: &str) -> Identity {
+        Identity {
+            id: id.to_string(),
+            scopes: vec![],
+            resources: HashMap::new(),
+        }
+    }
+
+    fn resource_spec(
+        name: &str,
+        acl: AccessControl,
+        resource_id_path: Option<&str>,
+    ) -> OperationSpec {
+        OperationSpec::new(
+            name,
+            OperationType::Query,
+            Visibility::External,
+            serde_json::json!({}),
+            serde_json::json!({}),
+            vec![],
+            acl,
+            resource_id_path.map(|s| s.to_string()),
+        )
+    }
+
+    #[test]
+    fn extract_json_pointer_resolves_single_field() {
+        let input = serde_json::json!({"containerId": "abc123"});
+        assert_eq!(
+            extract_json_pointer(&input, "$.containerId"),
+            Some("abc123".to_string())
+        );
+    }
+
+    #[test]
+    fn extract_json_pointer_missing_field_returns_none() {
+        let input = serde_json::json!({"other": 1});
+        assert_eq!(extract_json_pointer(&input, "$.containerId"), None);
+    }
+
+    #[test]
+    fn extract_json_pointer_non_string_value_returns_none() {
+        let input = serde_json::json!({"containerId": 42});
+        assert_eq!(extract_json_pointer(&input, "$.containerId"), None);
+    }
+
+    #[test]
+    fn extract_json_pointer_no_leading_dollar_returns_none() {
+        let input = serde_json::json!({"containerId": "abc"});
+        assert_eq!(extract_json_pointer(&input, "containerId"), None);
+    }
+
+    #[test]
+    fn extract_json_pointer_empty_after_dollar_returns_none() {
+        let input = serde_json::json!({"a": "b"});
+        assert_eq!(extract_json_pointer(&input, "$"), None);
+    }
+
+    #[test]
+    fn extract_json_pointer_nested_slash_path() {
+        let input = serde_json::json!({"data": {"id": "xyz"}});
+        assert_eq!(
+            extract_json_pointer(&input, "$.data/id"),
+            Some("xyz".to_string())
+        );
+    }
+
+    #[test]
+    fn extract_json_pointer_nested_dotted_path() {
+        let input = serde_json::json!({"data": {"id": "xyz"}});
+        assert_eq!(
+            extract_json_pointer(&input, "$.data.id"),
+            Some("xyz".to_string())
+        );
+    }
+
+    #[tokio::test]
+    async fn invoke_with_ownership_provider_allows_owned_resource() {
+        let mut registry = OperationRegistry::new();
+        let acl = AccessControl {
+            resource_type: Some("container".to_string()),
+            resource_action: Some("exec".to_string()),
+            ..Default::default()
+        };
+        registry
+            .register(HandlerRegistration::new(
+                resource_spec("container/exec", acl, Some("$.containerId")),
+                HandlerKind::Once(echo_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let provider: Arc<dyn OwnershipProvider> = Arc::new(MockOwnership {
+            owned: vec![("container".to_string(), "c1".to_string())],
+        });
+        let ctx = root_context_with_ownership(
+            "req-owns",
+            Some(empty_identity("alice")),
+            Some(provider),
+            ScopedPeerEnv::empty(),
+        );
+        let response = registry
+            .invoke(
+                "container/exec",
+                serde_json::json!({"containerId": "c1"}),
+                ctx,
+            )
+            .await;
+        assert!(response.result.is_ok(), "owned resource should be allowed");
+    }
+
+    #[tokio::test]
+    async fn invoke_with_ownership_provider_forbids_unowned_resource() {
+        let mut registry = OperationRegistry::new();
+        let acl = AccessControl {
+            resource_type: Some("container".to_string()),
+            resource_action: Some("exec".to_string()),
+            ..Default::default()
+        };
+        registry
+            .register(HandlerRegistration::new(
+                resource_spec("container/exec", acl, Some("$.containerId")),
+                HandlerKind::Once(echo_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let provider: Arc<dyn OwnershipProvider> = Arc::new(MockOwnership {
+            owned: vec![("container".to_string(), "c1".to_string())],
+        });
+        let ctx = root_context_with_ownership(
+            "req-not-owns",
+            Some(empty_identity("alice")),
+            Some(provider),
+            ScopedPeerEnv::empty(),
+        );
+        let response = registry
+            .invoke(
+                "container/exec",
+                serde_json::json!({"containerId": "c2"}),
+                ctx,
+            )
+            .await;
+        match response.result {
+            Err(e) => {
+                assert_eq!(e.code, "FORBIDDEN");
+                assert!(e.message.contains("container/c2"));
+            }
+            other => panic!("expected FORBIDDEN, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn invoke_with_ownership_provider_missing_field_falls_back_to_owns_any() {
+        let mut registry = OperationRegistry::new();
+        let acl = AccessControl {
+            resource_type: Some("container".to_string()),
+            resource_action: Some("exec".to_string()),
+            ..Default::default()
+        };
+        registry
+            .register(HandlerRegistration::new(
+                resource_spec("container/exec", acl, Some("$.containerId")),
+                HandlerKind::Once(echo_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let provider: Arc<dyn OwnershipProvider> = Arc::new(MockOwnership {
+            owned: vec![("container".to_string(), "c1".to_string())],
+        });
+        let ctx = root_context_with_ownership(
+            "req-missing",
+            Some(empty_identity("alice")),
+            Some(provider),
+            ScopedPeerEnv::empty(),
+        );
+        let response = registry
+            .invoke("container/exec", serde_json::json!({}), ctx)
+            .await;
+        assert!(
+            response.result.is_ok(),
+            "missing field → resource_id None → owns_any path allowed"
+        );
+    }
+
+    #[tokio::test]
+    async fn invoke_with_ownership_provider_missing_field_forbids_when_not_owns_any() {
+        let mut registry = OperationRegistry::new();
+        let acl = AccessControl {
+            resource_type: Some("container".to_string()),
+            resource_action: Some("exec".to_string()),
+            ..Default::default()
+        };
+        registry
+            .register(HandlerRegistration::new(
+                resource_spec("container/exec", acl, Some("$.containerId")),
+                HandlerKind::Once(echo_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let provider: Arc<dyn OwnershipProvider> = Arc::new(MockOwnership {
+            owned: vec![("volume".to_string(), "v1".to_string())],
+        });
+        let ctx = root_context_with_ownership(
+            "req-missing-denied",
+            Some(empty_identity("alice")),
+            Some(provider),
+            ScopedPeerEnv::empty(),
+        );
+        let response = registry
+            .invoke("container/exec", serde_json::json!({}), ctx)
+            .await;
+        match response.result {
+            Err(e) => {
+                assert_eq!(e.code, "FORBIDDEN");
+                assert!(e.message.contains("container"));
+            }
+            other => panic!("expected FORBIDDEN, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn invoke_without_ownership_provider_falls_back_to_static_resources() {
+        let mut registry = OperationRegistry::new();
+        let acl = AccessControl {
+            resource_type: Some("container".to_string()),
+            resource_action: Some("exec".to_string()),
+            ..Default::default()
+        };
+        registry
+            .register(HandlerRegistration::new(
+                resource_spec("container/exec", acl, Some("$.containerId")),
+                HandlerKind::Once(echo_handler()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .unwrap();
+        let mut resources = HashMap::new();
+        resources.insert("container".to_string(), vec!["exec".to_string()]);
+        let identity = Identity {
+            id: "alice".to_string(),
+            scopes: vec![],
+            resources,
+        };
+        let ctx =
+            root_context_with_ownership("req-static", Some(identity), None, ScopedPeerEnv::empty());
+        let response = registry
+            .invoke(
+                "container/exec",
+                serde_json::json!({"containerId": "c1"}),
+                ctx,
+            )
+            .await;
+        assert!(
+            response.result.is_ok(),
+            "no provider wired → static Identity.resources fallback allows"
+        );
+    }
 }

crates/alknet-call/src/registry/spec.rs

@@ -5,6 +5,7 @@
 //! specification.
 
 use alknet_core::auth::Identity;
+use alknet_core::ownership::OwnershipProvider;
 use serde_json::Value;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -56,7 +57,12 @@ impl AccessControl {
             || self.resource_action.is_some()
     }
 
-    pub fn check(&self, identity: Option<&Identity>) -> AccessResult {
+    pub fn check(
+        &self,
+        identity: Option<&Identity>,
+        resource_id: Option<&str>,
+        ownership: Option<&dyn OwnershipProvider>,
+    ) -> AccessResult {
         if !self.has_restrictions() {
             return AccessResult::Allowed;
         }
@@ -80,6 +86,29 @@ impl AccessControl {
             }
         }
 
+        if let Some(p) = ownership {
+            if let Some(rt) = &self.resource_type {
+                match resource_id {
+                    Some(rid) => {
+                        let action = self.resource_action.as_deref().unwrap_or("");
+                        if !p.owns(identity, rt, rid, action) {
+                            return AccessResult::Forbidden(format!(
+                                "not owner of resource: {rt}/{rid}"
+                            ));
+                        }
+                    }
+                    None => {
+                        if !p.owns_any(identity, rt) {
+                            return AccessResult::Forbidden(format!(
+                                "no owned resources of type: {rt}"
+                            ));
+                        }
+                    }
+                }
+                return AccessResult::Allowed;
+            }
+        }
+
         if let Some(rt) = &self.resource_type {
             let allowed = identity.resources.get(rt);
             match &self.resource_action {
@@ -118,9 +147,17 @@ pub struct OperationSpec {
     pub output_schema: Value,
     pub error_schemas: Vec<ErrorDefinition>,
     pub access_control: AccessControl,
+    /// JSON pointer into the input for the resource ID, when
+    /// `access_control.resource_type` is set and the operation targets a
+    /// specific runtime-spawned resource (ADR-050). e.g. `"$.containerId"`
+    /// for `docker/container/exec`. Absent for no-specific-resource
+    /// operations (the `list` case). `None` for operations with no
+    /// `resource_type` or with static resource sets.
+    pub resource_id_path: Option<String>,
 }
 
 impl OperationSpec {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         name: impl Into<String>,
         op_type: OperationType,
@@ -129,6 +166,7 @@ impl OperationSpec {
         output_schema: Value,
         error_schemas: Vec<ErrorDefinition>,
         access_control: AccessControl,
+        resource_id_path: Option<String>,
     ) -> Self {
         let name = name.into();
         let namespace = name
@@ -146,6 +184,7 @@ impl OperationSpec {
             output_schema,
             error_schemas,
             access_control,
+            resource_id_path,
         }
     }
 
@@ -184,6 +223,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             AccessControl::default(),
+            None,
         );
         assert_eq!(spec.path(), "/fs/readFile");
     }
@@ -198,6 +238,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             AccessControl::default(),
+            None,
         );
         assert_eq!(spec.namespace, "agent");
         assert_eq!(spec.name, "agent/chat");
@@ -213,16 +254,32 @@ mod tests {
             serde_json::json!({}),
             vec![],
             AccessControl::default(),
+            None,
         );
         assert_eq!(spec.namespace, "list");
     }
 
+    #[test]
+    fn resource_id_path_defaults_to_none() {
+        let spec = OperationSpec::new(
+            "fs/readFile",
+            OperationType::Query,
+            Visibility::External,
+            serde_json::json!({}),
+            serde_json::json!({}),
+            vec![],
+            AccessControl::default(),
+            None,
+        );
+        assert_eq!(spec.resource_id_path, None);
+    }
+
     #[test]
     fn empty_access_control_allowed_for_all() {
         let acl = AccessControl::default();
-        assert_eq!(acl.check(None), AccessResult::Allowed);
+        assert_eq!(acl.check(None, None, None), AccessResult::Allowed);
         let id = identity(&[], &[]);
-        assert_eq!(acl.check(Some(&id)), AccessResult::Allowed);
+        assert_eq!(acl.check(Some(&id), None, None), AccessResult::Allowed);
     }
 
     #[test]
@@ -232,7 +289,7 @@ mod tests {
             ..Default::default()
         };
         assert_eq!(
-            acl.check(None),
+            acl.check(None, None, None),
             AccessResult::Forbidden("authentication required".to_string())
         );
 
@@ -241,7 +298,7 @@ mod tests {
             ..Default::default()
         };
         assert_eq!(
-            acl2.check(None),
+            acl2.check(None, None, None),
             AccessResult::Forbidden("authentication required".to_string())
         );
 
@@ -250,7 +307,7 @@ mod tests {
             ..Default::default()
         };
         assert_eq!(
-            acl3.check(None),
+            acl3.check(None, None, None),
             AccessResult::Forbidden("authentication required".to_string())
         );
     }
@@ -263,11 +320,11 @@ mod tests {
         };
         let id_missing = identity(&["a"], &[]);
         assert!(matches!(
-            acl.check(Some(&id_missing)),
+            acl.check(Some(&id_missing), None, None),
             AccessResult::Forbidden(_)
         ));
         let id_ok = identity(&["a", "b", "c"], &[]);
-        assert_eq!(acl.check(Some(&id_ok)), AccessResult::Allowed);
+        assert_eq!(acl.check(Some(&id_ok), None, None), AccessResult::Allowed);
     }
 
     #[test]
@@ -277,12 +334,12 @@ mod tests {
             ..Default::default()
         };
         let id_x = identity(&["x"], &[]);
-        assert_eq!(acl.check(Some(&id_x)), AccessResult::Allowed);
+        assert_eq!(acl.check(Some(&id_x), None, None), AccessResult::Allowed);
         let id_y = identity(&["y"], &[]);
-        assert_eq!(acl.check(Some(&id_y)), AccessResult::Allowed);
+        assert_eq!(acl.check(Some(&id_y), None, None), AccessResult::Allowed);
         let id_none = identity(&["z"], &[]);
         assert!(matches!(
-            acl.check(Some(&id_none)),
+            acl.check(Some(&id_none), None, None),
             AccessResult::Forbidden(_)
         ));
     }
@@ -295,15 +352,15 @@ mod tests {
             ..Default::default()
         };
         let id_ok = identity(&[], &[("service", &["read"])]);
-        assert_eq!(acl.check(Some(&id_ok)), AccessResult::Allowed);
+        assert_eq!(acl.check(Some(&id_ok), None, None), AccessResult::Allowed);
         let id_missing_action = identity(&[], &[("service", &["write"])]);
         assert!(matches!(
-            acl.check(Some(&id_missing_action)),
+            acl.check(Some(&id_missing_action), None, None),
             AccessResult::Forbidden(_)
         ));
         let id_missing_type = identity(&[], &[("other", &["read"])]);
         assert!(matches!(
-            acl.check(Some(&id_missing_type)),
+            acl.check(Some(&id_missing_type), None, None),
             AccessResult::Forbidden(_)
         ));
     }
@@ -317,10 +374,156 @@ mod tests {
             ..Default::default()
         };
         let id_ok = identity(&["admin"], &[("service", &["read"])]);
-        assert_eq!(acl.check(Some(&id_ok)), AccessResult::Allowed);
+        assert_eq!(acl.check(Some(&id_ok), None, None), AccessResult::Allowed);
         let id_missing_scope = identity(&["user"], &[("service", &["read"])]);
         assert!(matches!(
-            acl.check(Some(&id_missing_scope)),
+            acl.check(Some(&id_missing_scope), None, None),
+            AccessResult::Forbidden(_)
+        ));
+    }
+
+    struct MockOwnership {
+        owned: Vec<(String, String)>,
+    }
+
+    impl OwnershipProvider for MockOwnership {
+        fn owns(
+            &self,
+            _identity: &Identity,
+            resource_type: &str,
+            resource_id: &str,
+            _action: &str,
+        ) -> bool {
+            self.owned
+                .iter()
+                .any(|(rt, rid)| rt == resource_type && rid == resource_id)
+        }
+
+        fn owned_resources(&self, _identity: &Identity, resource_type: &str) -> Vec<String> {
+            self.owned
+                .iter()
+                .filter(|(rt, _)| rt == resource_type)
+                .map(|(_, rid)| rid.clone())
+                .collect()
+        }
+
+        fn owns_any(&self, _identity: &Identity, resource_type: &str) -> bool {
+            self.owned.iter().any(|(rt, _)| rt == resource_type)
+        }
+    }
+
+    fn empty_identity(id: &str) -> Identity {
+        Identity {
+            id: id.to_string(),
+            scopes: vec![],
+            resources: HashMap::new(),
+        }
+    }
+
+    #[test]
+    fn ownership_provider_allows_owned_resource() {
+        let acl = AccessControl {
+            resource_type: Some("container".to_string()),
+            resource_action: Some("exec".to_string()),
+            ..Default::default()
+        };
+        let id = empty_identity("alice");
+        let provider = MockOwnership {
+            owned: vec![("container".to_string(), "c1".to_string())],
+        };
+        assert_eq!(
+            acl.check(
+                Some(&id),
+                Some("c1"),
+                Some(&provider as &dyn OwnershipProvider)
+            ),
+            AccessResult::Allowed
+        );
+    }
+
+    #[test]
+    fn ownership_provider_forbids_unowned_resource() {
+        let acl = AccessControl {
+            resource_type: Some("container".to_string()),
+            resource_action: Some("exec".to_string()),
+            ..Default::default()
+        };
+        let id = empty_identity("alice");
+        let provider = MockOwnership {
+            owned: vec![("container".to_string(), "c1".to_string())],
+        };
+        assert!(matches!(
+            acl.check(
+                Some(&id),
+                Some("c2"),
+                Some(&provider as &dyn OwnershipProvider)
+            ),
+            AccessResult::Forbidden(_)
+        ));
+    }
+
+    #[test]
+    fn ownership_provider_forbids_none_identity() {
+        let acl = AccessControl {
+            resource_type: Some("container".to_string()),
+            resource_action: Some("exec".to_string()),
+            ..Default::default()
+        };
+        let provider = MockOwnership {
+            owned: vec![("container".to_string(), "c1".to_string())],
+        };
+        assert!(matches!(
+            acl.check(None, Some("c1"), Some(&provider as &dyn OwnershipProvider)),
+            AccessResult::Forbidden(_)
+        ));
+    }
+
+    #[test]
+    fn ownership_provider_list_allowed_when_owns_any() {
+        let acl = AccessControl {
+            resource_type: Some("container".to_string()),
+            resource_action: Some("exec".to_string()),
+            ..Default::default()
+        };
+        let id = empty_identity("alice");
+        let provider = MockOwnership {
+            owned: vec![("container".to_string(), "c1".to_string())],
+        };
+        assert_eq!(
+            acl.check(Some(&id), None, Some(&provider as &dyn OwnershipProvider)),
+            AccessResult::Allowed
+        );
+    }
+
+    #[test]
+    fn ownership_provider_list_forbidden_when_not_owns_any() {
+        let acl = AccessControl {
+            resource_type: Some("container".to_string()),
+            resource_action: Some("exec".to_string()),
+            ..Default::default()
+        };
+        let id = empty_identity("alice");
+        let provider = MockOwnership {
+            owned: vec![("volume".to_string(), "v1".to_string())],
+        };
+        assert!(matches!(
+            acl.check(Some(&id), None, Some(&provider as &dyn OwnershipProvider)),
+            AccessResult::Forbidden(_)
+        ));
+    }
+
+    #[test]
+    fn ownership_none_falls_back_to_static() {
+        let acl = AccessControl {
+            resource_type: Some("service".to_string()),
+            resource_action: Some("read".to_string()),
+            ..Default::default()
+        };
+        let id_ok = identity(&[], &[("service", &["read"])]);
+        assert_eq!(acl.check(Some(&id_ok), None, None), AccessResult::Allowed);
+        let id_missing = identity(&[], &[("service", &["write"])]);
+        assert!(matches!(
+            acl.check(Some(&id_missing), None, None),
             AccessResult::Forbidden(_)
         ));
     }

crates/alknet-call/tests/two_node_call.rs

@@ -40,6 +40,7 @@ fn external_spec(name: &str) -> OperationSpec {
         serde_json::json!({}),
         vec![],
         AccessControl::default(),
+        None,
     )
 }
 
@@ -308,6 +309,7 @@ async fn from_call_discovers_and_forwards_over_quic_loopback() {
         abort_policy: alknet_call::registry::context::AbortPolicy::default(),
         deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
         internal: true,
+        ownership: None,
     };
 
     let response = tokio::time::timeout(

crates/alknet-core/src/lib.rs

@@ -10,8 +10,10 @@ pub mod auth;
 pub mod config;
 pub mod endpoint;
 pub mod fingerprint;
+pub mod ownership;
 pub mod store;
 pub mod types;
 
 pub use auth::{IdentityProvider, IdentityStore};
+pub use ownership::{InMemoryOwnershipStore, OwnershipError, OwnershipProvider, OwnershipStore};
 pub use store::{CredentialStore, EncryptedData, InMemoryCredentialStore, StoreError};

crates/alknet-core/src/ownership.rs

@@ -0,0 +1,280 @@
+//! Ownership store: `OwnershipProvider` (sync read trait), `OwnershipStore`
+//! (async write trait), `InMemoryOwnershipStore` default adapter, and
+//! `OwnershipError`.
+//!
+//! See `docs/architecture/crates/core/auth.md` and ADR-050 for the full
+//! specification. Runtime-spawned resources (containers, TTYs, workspace
+//! processes) have derived ownership — whoever spawned the resource owns it.
+//! The static `Identity.resources` model can't represent this (the resource
+//! didn't exist when the identity was resolved), so `AccessControl::check`
+//! (in alknet-call) consults `OwnershipProvider` at check time. This is the
+//! fourth instance of the repo/adapter pattern (ADR-033).
+
+use std::collections::HashMap;
+use std::sync::RwLock;
+
+use async_trait::async_trait;
+
+use crate::auth::Identity;
+
+#[non_exhaustive]
+#[derive(Debug, thiserror::Error)]
+pub enum OwnershipError {
+    #[error("backend error: {message}")]
+    Backend { message: String },
+    #[error("not found: {entity}")]
+    NotFound { entity: String },
+}
+
+/// Read side: consulted by `AccessControl::check` on the dispatch hot path.
+/// Sync — called in the dispatch loop, no `.await`.
+pub trait OwnershipProvider: Send + Sync + 'static {
+    /// Does `identity` own `resource_type/resource_id` with `action`?
+    /// The `action` parameter is accepted but not gated on — the base model
+    /// is "owner can do anything they own." Per-action grants are a future
+    /// extension; this preserves the door without building the mechanism.
+    fn owns(
+        &self,
+        identity: &Identity,
+        resource_type: &str,
+        resource_id: &str,
+        action: &str,
+    ) -> bool;
+
+    /// What resources of `resource_type` does `identity` own? Returns the
+    /// set of resource IDs the caller owns, for the handler to filter
+    /// against (the result-filter path, ADR-050 §4a).
+    fn owned_resources(&self, identity: &Identity, resource_type: &str) -> Vec<String>;
+
+    /// Does `identity` own *any* resource of `resource_type`? The scope-gate
+    /// path (ADR-050 §4a).
+    fn owns_any(&self, identity: &Identity, resource_type: &str) -> bool;
+}
+
+/// Write side: called by the handler that manages the resource lifecycle.
+/// Async — not on the dispatch hot path. The handler calls `record` on
+/// spawn and `revoke` on teardown (ADR-050 §4b — handler-driven, not a
+/// reaper). The trait takes `&self` so it can be shared as
+/// `Arc<dyn OwnershipStore>` (interior mutability via `RwLock`).
+#[async_trait]
+pub trait OwnershipStore: Send + Sync + 'static {
+    /// Record that `identity` spawned `resource_type/resource_id`.
+    async fn record(
+        &self,
+        identity: &Identity,
+        resource_type: &str,
+        resource_id: &str,
+    ) -> Result<(), OwnershipError>;
+
+    /// Revoke ownership of `resource_type/resource_id`. Called by the
+    /// handler on resource teardown (ADR-050 §4b).
+    async fn revoke(&self, resource_type: &str, resource_id: &str) -> Result<(), OwnershipError>;
+}
+
+pub struct InMemoryOwnershipStore {
+    inner: RwLock<HashMap<(String, String), Identity>>,
+}
+
+impl InMemoryOwnershipStore {
+    pub fn new() -> Self {
+        Self {
+            inner: RwLock::new(HashMap::new()),
+        }
+    }
+}
+
+impl Default for InMemoryOwnershipStore {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl OwnershipProvider for InMemoryOwnershipStore {
+    fn owns(
+        &self,
+        identity: &Identity,
+        resource_type: &str,
+        resource_id: &str,
+        _action: &str,
+    ) -> bool {
+        let inner = self.inner.read().unwrap_or_else(|e| e.into_inner());
+        inner
+            .get(&(resource_type.to_string(), resource_id.to_string()))
+            .map(|owner| owner.id == identity.id)
+            .unwrap_or(false)
+    }
+
+    fn owned_resources(&self, identity: &Identity, resource_type: &str) -> Vec<String> {
+        let inner = self.inner.read().unwrap_or_else(|e| e.into_inner());
+        inner
+            .iter()
+            .filter(|((rt, _), owner)| rt == resource_type && owner.id == identity.id)
+            .map(|((_, rid), _)| rid.clone())
+            .collect()
+    }
+
+    fn owns_any(&self, identity: &Identity, resource_type: &str) -> bool {
+        let inner = self.inner.read().unwrap_or_else(|e| e.into_inner());
+        inner
+            .iter()
+            .any(|((rt, _), owner)| rt == resource_type && owner.id == identity.id)
+    }
+}
+
+#[async_trait]
+impl OwnershipStore for InMemoryOwnershipStore {
+    async fn record(
+        &self,
+        identity: &Identity,
+        resource_type: &str,
+        resource_id: &str,
+    ) -> Result<(), OwnershipError> {
+        let mut inner = self.inner.write().unwrap_or_else(|e| e.into_inner());
+        inner.insert(
+            (resource_type.to_string(), resource_id.to_string()),
+            identity.clone(),
+        );
+        Ok(())
+    }
+
+    /// Revoking a non-existent resource is a no-op: returns `Ok(())`. This
+    /// mirrors `InMemoryCredentialStore::delete`, which treats a missing
+    /// entry as success. Teardown paths are idempotent — a handler may call
+    /// `revoke` on a resource that was already removed (e.g. crashed and
+    /// re-entered cleanup), and should not error in that case.
+    async fn revoke(&self, resource_type: &str, resource_id: &str) -> Result<(), OwnershipError> {
+        let mut inner = self.inner.write().unwrap_or_else(|e| e.into_inner());
+        inner.remove(&(resource_type.to_string(), resource_id.to_string()));
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn make_identity(id: &str) -> Identity {
+        Identity {
+            id: id.to_string(),
+            scopes: vec![],
+            resources: HashMap::new(),
+        }
+    }
+
+    #[tokio::test]
+    async fn record_owns_revoke_round_trip() {
+        let store = InMemoryOwnershipStore::new();
+        let owner = make_identity("worker-a");
+
+        assert!(!store.owns(&owner, "container", "c1", "exec"));
+        assert!(!store.owns_any(&owner, "container"));
+        assert!(store.owned_resources(&owner, "container").is_empty());
+
+        store.record(&owner, "container", "c1").await.unwrap();
+        assert!(store.owns(&owner, "container", "c1", "exec"));
+        assert!(store.owns(&owner, "container", "c1", "logs"));
+        assert!(store.owns_any(&owner, "container"));
+        assert_eq!(store.owned_resources(&owner, "container"), vec!["c1"]);
+
+        store.revoke("container", "c1").await.unwrap();
+        assert!(!store.owns(&owner, "container", "c1", "exec"));
+        assert!(!store.owns_any(&owner, "container"));
+        assert!(store.owned_resources(&owner, "container").is_empty());
+    }
+
+    #[tokio::test]
+    async fn owned_resources_returns_all_for_owner_with_multiple() {
+        let store = InMemoryOwnershipStore::new();
+        let owner = make_identity("worker-a");
+        store.record(&owner, "container", "c1").await.unwrap();
+        store.record(&owner, "container", "c2").await.unwrap();
+        store.record(&owner, "container", "c3").await.unwrap();
+
+        let mut owned = store.owned_resources(&owner, "container");
+        owned.sort();
+        assert_eq!(owned, vec!["c1", "c2", "c3"]);
+    }
+
+    #[tokio::test]
+    async fn owned_resources_filters_by_resource_type() {
+        let store = InMemoryOwnershipStore::new();
+        let owner = make_identity("worker-a");
+        store.record(&owner, "container", "c1").await.unwrap();
+        store.record(&owner, "tty", "t1").await.unwrap();
+
+        let owned_containers = store.owned_resources(&owner, "container");
+        assert_eq!(owned_containers, vec!["c1"]);
+        let owned_ttys = store.owned_resources(&owner, "tty");
+        assert_eq!(owned_ttys, vec!["t1"]);
+    }
+
+    #[tokio::test]
+    async fn owns_any_returns_false_for_owner_with_no_resources_of_type() {
+        let store = InMemoryOwnershipStore::new();
+        let owner = make_identity("worker-a");
+        store.record(&owner, "container", "c1").await.unwrap();
+
+        assert!(store.owns_any(&owner, "container"));
+        assert!(!store.owns_any(&owner, "tty"));
+    }
+
+    #[tokio::test]
+    async fn revoke_on_non_existent_resource_is_no_op() {
+        let store = InMemoryOwnershipStore::new();
+        store.revoke("container", "never-existed").await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn owns_returns_false_for_different_identity() {
+        let store = InMemoryOwnershipStore::new();
+        let owner = make_identity("worker-a");
+        let other = make_identity("worker-b");
+        store.record(&owner, "container", "c1").await.unwrap();
+
+        assert!(store.owns(&owner, "container", "c1", "exec"));
+        assert!(!store.owns(&other, "container", "c1", "exec"));
+        assert!(!store.owns_any(&other, "container"));
+        assert!(store.owned_resources(&other, "container").is_empty());
+    }
+
+    #[tokio::test]
+    async fn record_replaces_existing_owner() {
+        let store = InMemoryOwnershipStore::new();
+        let owner_a = make_identity("worker-a");
+        let owner_b = make_identity("worker-b");
+        store.record(&owner_a, "container", "c1").await.unwrap();
+        store.record(&owner_b, "container", "c1").await.unwrap();
+
+        assert!(!store.owns(&owner_a, "container", "c1", "exec"));
+        assert!(store.owns(&owner_b, "container", "c1", "exec"));
+    }
+
+    #[tokio::test]
+    async fn default_is_empty_store() {
+        let store = InMemoryOwnershipStore::default();
+        let owner = make_identity("worker-a");
+        assert!(store.owned_resources(&owner, "container").is_empty());
+        assert!(!store.owns_any(&owner, "container"));
+    }
+
+    #[test]
+    fn ownership_error_display_formatting() {
+        let backend = OwnershipError::Backend {
+            message: "disk full".to_string(),
+        };
+        assert_eq!(backend.to_string(), "backend error: disk full");
+
+        let not_found = OwnershipError::NotFound {
+            entity: "container:c1".to_string(),
+        };
+        assert_eq!(not_found.to_string(), "not found: container:c1");
+    }
+
+    #[test]
+    fn ownership_error_is_non_exhaustive() {
+        let err = OwnershipError::Backend {
+            message: "x".to_string(),
+        };
+        let _ = err.to_string();
+    }
+}

crates/alknet-http/src/adapters/from_mcp/mod.rs

@@ -180,6 +180,7 @@ pub(crate) fn build_spec(tool: &Tool, namespace: &str) -> OperationSpec {
         output_schema,
         error_schemas,
         AccessControl::default(),
+        None,
     )
 }
 

crates/alknet-http/src/adapters/from_openapi.rs

@@ -427,6 +427,7 @@ impl FromOpenAPI {
             output_schema,
             error_schemas,
             AccessControl::default(),
+            None,
         );
 
         let path_template = path.to_string();
@@ -967,6 +968,7 @@ mod tests {
             abort_policy: alknet_call::registry::context::AbortPolicy::default(),
             deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
             internal: true,
+            ownership: None,
         }
     }
 

crates/alknet-http/src/adapters/to_mcp.rs

@@ -488,6 +488,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             acl,
+            None,
         )
     }
 

crates/alknet-http/src/adapters/to_openapi.rs

@@ -560,6 +560,7 @@ mod tests {
             json!({}),
             errors,
             AccessControl::default(),
+            None,
         )
     }
 
@@ -1015,6 +1016,7 @@ mod tests {
                     json!({}),
                     vec![error("INTERNAL_ERROR", Some(418))],
                     AccessControl::default(),
+                    None,
                 ),
                 HandlerKind::Once(noop_handler()),
                 OperationProvenance::Local,

crates/alknet-http/src/gateway/dispatch.rs

@@ -135,6 +135,7 @@ impl GatewayDispatch {
             env,
             abort_policy: AbortPolicy::default(),
             internal: false,
+            ownership: None,
         }
     }
 }
@@ -204,6 +205,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             acl,
+            None,
         )
     }
 
@@ -216,6 +218,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             acl,
+            None,
         )
     }
 
@@ -231,6 +234,7 @@ mod tests {
                     serde_json::json!({}),
                     vec![],
                     acl,
+                    None,
                 ),
                 HandlerKind::Once(make_handler(|input, context| async move {
                     ResponseEnvelope::ok(context.request_id, input)
@@ -278,6 +282,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             acl,
+            None,
         )
     }
 

crates/alknet-http/src/server/gateway_routes.rs

@@ -261,7 +261,7 @@ fn access_check_for_op(
 ) -> Option<String> {
     let name = operation.strip_prefix('/').unwrap_or(operation);
     let reg = registry.registration(name)?;
-    if let AccessResult::Forbidden(message) = reg.spec.access_control.check(identity) {
+    if let AccessResult::Forbidden(message) = reg.spec.access_control.check(identity, None, None) {
         return Some(message);
     }
     None
@@ -349,6 +349,7 @@ mod tests {
             json!({}),
             vec![],
             acl,
+            None,
         )
     }
 
@@ -361,6 +362,7 @@ mod tests {
             json!({}),
             vec![],
             AccessControl::default(),
+            None,
         )
     }
 
@@ -428,6 +430,7 @@ mod tests {
             json!({}),
             vec![],
             acl,
+            None,
         )
     }
 

crates/alknet-http/src/websocket/mod.rs

@@ -72,6 +72,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             AccessControl::default(),
+            None,
         )
     }
 

crates/alknet-http/src/websocket/overlay.rs

@@ -91,6 +91,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             acl,
+            None,
         )
     }
 
@@ -103,6 +104,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             AccessControl::default(),
+            None,
         )
     }
 
@@ -170,6 +172,7 @@ mod tests {
             abort_policy: AbortPolicy::default(),
             deadline: Some(Instant::now() + Duration::from_secs(30)),
             internal: true,
+            ownership: None,
         }
     }
 

crates/alknet-http/src/websocket/upgrade.rs

@@ -313,6 +313,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             acl,
+            None,
         )
     }
 
@@ -325,6 +326,7 @@ mod tests {
             serde_json::json!({}),
             vec![],
             AccessControl::default(),
+            None,
         )
     }
 
@@ -568,6 +570,7 @@ mod tests {
                     serde_json::json!({}),
                     vec![],
                     AccessControl::default(),
+                    None,
                 ),
                 HandlerKind::Once(make_handler(|input, ctx| async move {
                     ResponseEnvelope::ok(ctx.request_id, input)
@@ -928,6 +931,7 @@ mod tests {
             abort_policy: AbortPolicy::default(),
             deadline: Some(Instant::now() + Duration::from_secs(30)),
             internal: false,
+            ownership: None,
         }
     }
 
@@ -948,6 +952,7 @@ mod tests {
             abort_policy: AbortPolicy::default(),
             deadline: Some(Instant::now() + Duration::from_secs(30)),
             internal: true,
+            ownership: None,
         }
     }
 

crates/alknet-http/tests/from_mcp_integration.rs

@@ -62,6 +62,7 @@ fn test_context(request_id: &str, caps: Capabilities) -> OperationContext {
         abort_policy: AbortPolicy::default(),
         deadline: Some(Instant::now() + Duration::from_secs(30)),
         internal: true,
+        ownership: None,
     }
 }
 

tasks/call/registry/access-control-ownership-check.md

@@ -1,7 +1,7 @@
 ---
 id: call/registry/access-control-ownership-check
 name: Update AccessControl::check to consult OwnershipProvider for dynamic resource ownership (ADR-050 §2)
-status: pending
+status: done
 depends_on: [core/ownership-store-trait]
 scope: moderate
 risk: medium

tasks/call/registry/dispatch-resource-id-extraction.md

@@ -1,7 +1,7 @@
 ---
 id: call/registry/dispatch-resource-id-extraction
 name: Wire dispatch path to extract resource_id from input and thread OwnershipProvider to AccessControl::check (ADR-050 §2a, §4a)
-status: pending
+status: done
 depends_on: [call/registry/operation-spec-resource-id-path, call/registry/access-control-ownership-check]
 scope: moderate
 risk: medium

tasks/call/registry/operation-spec-resource-id-path.md

@@ -1,7 +1,7 @@
 ---
 id: call/registry/operation-spec-resource-id-path
 name: Add resource_id_path field to OperationSpec (ADR-050 §2a)
-status: pending
+status: done
 depends_on: []
 scope: single
 risk: low

tasks/core/ownership-store-trait.md

@@ -1,7 +1,7 @@
 ---
 id: core/ownership-store-trait
 name: Add OwnershipProvider (sync read) + OwnershipStore (async write) traits and InMemoryOwnershipStore (ADR-050)
-status: pending
+status: done
 depends_on: []
 scope: moderate
 risk: low