feat: add on_copy and on_patch optional lifecycle methods to Resource (PRA-286) (#26)

Add two new optional lifecycle methods to the provider resource interface
for subgraph copy and patch operations. Both methods are opt-in: existing
providers continue to work unchanged.

New Resource methods:
- on_copy(context) -> CopyResult: handles resource duplication with
  stateless/stateful copy strategies and tag propagation
- on_patch(patch) -> PatchResult: handles applying patches (migrations,
  schema diffs) with compatibility constraint matching
- supports_copy() / supports_patch(): classmethods for runtime detection
  of whether a resource type implements these operations

New models (in types.py):
- CopyStrategy, CopyContext, CopyResult
- CompatibilityConstraint, PatchDefinition, PatchResult

Also adds:
- COPY and PATCH to EventType enum
- invoke_copy() and invoke_patch() to ProviderHarness for local testing
- copy_result and patch_result fields on LifecycleResult
- 32 new tests covering all new functionality

Author: tngotr

src/pragma_sdk/__init__.py

@@ -56,14 +56,28 @@
 )
 from pragma_sdk.models import Provider as ProviderModel
 from pragma_sdk.provider import Provider
-from pragma_sdk.types import HealthStatus, LifecycleState, LogEntry
+from pragma_sdk.types import (
+    CompatibilityConstraint,
+    CopyContext,
+    CopyResult,
+    CopyStrategy,
+    HealthStatus,
+    LifecycleState,
+    LogEntry,
+    PatchDefinition,
+    PatchResult,
+)
 
 
 __all__ = [
     "AsyncPragmaClient",
     "BuildInfo",
     "BuildStatus",
+    "CompatibilityConstraint",
     "Config",
+    "CopyContext",
+    "CopyResult",
+    "CopyStrategy",
     "Dependency",
     "DeploymentResult",
     "DeploymentStatus",
@@ -80,6 +94,8 @@
     "OrganizationStatus",
     "Outputs",
     "PaginatedResponse",
+    "PatchDefinition",
+    "PatchResult",
     "PragmaClient",
     "Provider",
     "ProviderAuthor",

src/pragma_sdk/models/base.py

@@ -24,7 +24,15 @@
     Sensitive,
     format_resource_id,
 )
-from pragma_sdk.types import HealthStatus, LifecycleState, LogEntry
+from pragma_sdk.types import (
+    CopyContext,
+    CopyResult,
+    HealthStatus,
+    LifecycleState,
+    LogEntry,
+    PatchDefinition,
+    PatchResult,
+)
 
 
 def _is_union_origin(origin: Any) -> bool:
@@ -714,6 +722,70 @@ async def on_delete(self) -> None:
         """Handle resource deletion."""
         raise NotImplementedError(f"{self.__class__.__name__} must implement on_delete()")
 
+    async def on_copy(self, context: CopyContext) -> CopyResult:
+        """Handle resource duplication as part of a subgraph copy operation.
+
+        Override this method to control how the resource duplicates itself.
+        Stateless resources (config-only) can return a modified config directly.
+        Stateful resources (data-bearing) may need to clone underlying data
+        (e.g., create a new database, copy vector indices).
+
+        The default implementation raises NotImplementedError, indicating the
+        resource type does not support copying. Use ``supports_copy()`` to check
+        before calling.
+
+        Args:
+            context: Copy context including tags, target name, strategy, and
+                provider-specific metadata.
+
+        Returns:
+            CopyResult with the configuration for the new copied resource.
+
+        Raises:
+            NotImplementedError: If the resource type does not support copying.
+        """
+        raise NotImplementedError(f"{self.__class__.__name__} does not support on_copy()")
+
+    async def on_patch(self, patch: PatchDefinition) -> PatchResult:
+        """Handle applying a patch (migration script, schema diff) to this resource.
+
+        Override this method to control how patches are applied to the resource.
+        Patches have compatibility constraints that are checked before this method
+        is called -- the resource is guaranteed to satisfy all constraints.
+
+        The default implementation raises NotImplementedError, indicating the
+        resource type does not support patching. Use ``supports_patch()`` to check
+        before calling.
+
+        Args:
+            patch: Patch definition including the payload, constraints, and metadata.
+
+        Returns:
+            PatchResult indicating success/failure and any modified config/outputs.
+
+        Raises:
+            NotImplementedError: If the resource type does not support patching.
+        """
+        raise NotImplementedError(f"{self.__class__.__name__} does not support on_patch()")
+
+    @classmethod
+    def supports_copy(cls) -> bool:
+        """Check if this resource type implements on_copy.
+
+        Returns:
+            True if the resource subclass overrides on_copy from the base Resource.
+        """
+        return cls.on_copy is not Resource.on_copy
+
+    @classmethod
+    def supports_patch(cls) -> bool:
+        """Check if this resource type implements on_patch.
+
+        Returns:
+            True if the resource subclass overrides on_patch from the base Resource.
+        """
+        return cls.on_patch is not Resource.on_patch
+
     @classmethod
     def upgrade(cls, config: dict, outputs: dict | None) -> tuple[dict, dict | None]:
         """Migrate config and outputs from the previous provider version.

src/pragma_sdk/models/enums.py

@@ -24,11 +24,13 @@ class DeploymentStatus(StrEnum):
 
 
 class EventType(StrEnum):
-    """Resource lifecycle event type: CREATE, UPDATE, or DELETE."""
+    """Resource lifecycle event type."""
 
     CREATE = "CREATE"
     UPDATE = "UPDATE"
     DELETE = "DELETE"
+    COPY = "COPY"
+    PATCH = "PATCH"
     MIGRATE_UP = "MIGRATE_UP"
     MIGRATE_DOWN = "MIGRATE_DOWN"
 

src/pragma_sdk/provider/harness.py

@@ -42,7 +42,7 @@ async def test_my_resource():
 
 from pragma_sdk.context import reset_provider_name, set_provider_name
 from pragma_sdk.models import Config, Outputs, Resource
-from pragma_sdk.types import LifecycleState
+from pragma_sdk.types import CopyContext, CopyResult, LifecycleState, PatchDefinition, PatchResult
 
 
 class EventType(StrEnum):
@@ -51,6 +51,8 @@ class EventType(StrEnum):
     CREATE = "create"
     UPDATE = "update"
     DELETE = "delete"
+    COPY = "copy"
+    PATCH = "patch"
 
 
 @dataclass
@@ -71,11 +73,14 @@ class LifecycleResult:
     """Result of executing a lifecycle method in tests.
 
     Use `.success` or `.failed` to check status, `.outputs` for returned values,
-    and `.error` for any exception raised.
+    and `.error` for any exception raised. For copy and patch operations, use
+    `.copy_result` and `.patch_result` respectively.
     """
 
     success: bool
     outputs: Outputs | None = None
+    copy_result: CopyResult | None = None
+    patch_result: PatchResult | None = None
     error: Exception | None = None
     resource: Resource | None = None
     event: LifecycleEvent | None = None
@@ -321,3 +326,125 @@ async def invoke_delete(
 
         self._results.append(result)
         return result
+
+    async def invoke_copy(
+        self,
+        resource_class: type[Resource],
+        name: str,
+        config: Config,
+        context: CopyContext,
+        current_outputs: Outputs | None = None,
+        tags: list[str] | None = None,
+    ) -> LifecycleResult:
+        """Invoke the on_copy lifecycle method.
+
+        Args:
+            resource_class: Resource subclass to test.
+            name: Source resource instance name.
+            config: Configuration of the source resource.
+            context: Copy context with target name, tags, strategy, and metadata.
+            current_outputs: Outputs of the source resource.
+            tags: Tags on the source resource.
+
+        Returns:
+            Result containing success status, copy result as outputs, and any error.
+        """
+        event = LifecycleEvent(
+            event_id=str(uuid4()),
+            event_type=EventType.COPY,
+            resource_class=resource_class,
+            name=name,
+            config=config,
+        )
+        self._events.append(event)
+
+        resource = resource_class(
+            name=name,
+            config=config,
+            lifecycle_state=LifecycleState.PROCESSING,
+            outputs=current_outputs,
+            tags=tags,
+        )
+
+        provider_token = set_provider_name(self._provider_name)
+        try:
+            copy_result = await resource.on_copy(context)
+            result = LifecycleResult(
+                success=True,
+                copy_result=copy_result,
+                resource=resource,
+                event=event,
+            )
+        except Exception as e:
+            result = LifecycleResult(
+                success=False,
+                error=e,
+                resource=resource,
+                event=event,
+            )
+        finally:
+            reset_provider_name(provider_token)
+
+        self._results.append(result)
+        return result
+
+    async def invoke_patch(
+        self,
+        resource_class: type[Resource],
+        name: str,
+        config: Config,
+        patch: PatchDefinition,
+        current_outputs: Outputs | None = None,
+        tags: list[str] | None = None,
+    ) -> LifecycleResult:
+        """Invoke the on_patch lifecycle method.
+
+        Args:
+            resource_class: Resource subclass to test.
+            name: Resource instance name.
+            config: Configuration of the resource.
+            patch: Patch definition to apply.
+            current_outputs: Current outputs of the resource.
+            tags: Tags on the resource.
+
+        Returns:
+            Result containing success status, patch result, and any error.
+        """
+        event = LifecycleEvent(
+            event_id=str(uuid4()),
+            event_type=EventType.PATCH,
+            resource_class=resource_class,
+            name=name,
+            config=config,
+        )
+        self._events.append(event)
+
+        resource = resource_class(
+            name=name,
+            config=config,
+            lifecycle_state=LifecycleState.PROCESSING,
+            outputs=current_outputs,
+            tags=tags,
+        )
+
+        provider_token = set_provider_name(self._provider_name)
+        try:
+            patch_result = await resource.on_patch(patch)
+            result = LifecycleResult(
+                success=True,
+                patch_result=patch_result,
+                resource=resource,
+                event=event,
+            )
+        except Exception as e:
+            result = LifecycleResult(
+                success=False,
+                error=e,
+                resource=resource,
+                event=event,
+            )
+        finally:
+            reset_provider_name(provider_token)
+
+        self._results.append(result)
+        return result