diff --git a/README.md b/README.md
index b24cb75..daacdd0 100644
--- a/README.md
+++ b/README.md
@@ -170,6 +170,105 @@ metadata = head_object_from_snapshot(s3_client, 'my-bucket', 'file.txt', '12345'
 rename_object(s3_client, 'my-bucket', 'old-name.txt', 'new-name.txt')
 ```
 
+### 4. Agent-storage workflows
+
+Higher-level helpers for AI-agent storage flows: workspaces (per-agent
+buckets), parallel forks, and checkpoints. Built on top of snapshots,
+forks, the standard S3 lifecycle API, and the AWS-IAM compatible Tigris IAM.
+
+#### Workspaces
+
+A workspace is a Tigris bucket dedicated to a single agent — created with
+snapshots enabled by default, with optional TTL for auto-cleanup and an
+optional bucket-scoped access key for least-privilege access.
+
+```python
+from tigris_boto3_ext import create_workspace, teardown_workspace
+
+ws = create_workspace(
+    s3_client,
+    'agent-abc',
+    ttl_days=1,                  # auto-expire objects after 1 day
+    credentials_role='Editor',   # provision a bucket-scoped access key
+)
+
+# ws.bucket — bucket name
+# ws.credentials.access_key_id / secret_access_key — scoped key (if requested)
+
+teardown_workspace(s3_client, ws)  # revokes credentials, force-deletes bucket
+```
+
+#### Forks (parallel agent runs)
+
+Snapshot a bucket, then fork it `count` times. Each fork is its own bucket,
+created instantly via copy-on-write — agents can read/write without affecting
+the base bucket or each other.
+
+```python
+from tigris_boto3_ext import create_forks, teardown_forks
+
+forks = create_forks(
+    s3_client,
+    'training-data',
+    count=3,
+    credentials_role='ReadOnly',  # one scoped key per fork (optional)
+)
+
+for fork in forks.forks:
+    print(fork.bucket)  # f"{base}-fork-{snapshot_id}-0", "-1", "-2"
+    # fork.credentials.access_key_id / secret_access_key (if requested)
+
+teardown_forks(s3_client, forks)
+```
+
+#### Checkpoints
+
+Capture a labeled snapshot you can later restore from. Restoring creates a new
+fork at that point in time; the original bucket is untouched.
+
+```python
+from tigris_boto3_ext import checkpoint, restore
+
+ck = checkpoint(s3_client, 'training-data', name='epoch-50')
+print(ck.snapshot_id)
+
+# Later — restore into a fresh fork named "training-data-restore-<snapshot_id>"
+restored_bucket = restore(s3_client, 'training-data', ck.snapshot_id)
+```
+
+To enumerate existing checkpoints on a bucket, use the lower-level
+`list_snapshots` and parse the `Name` field (`"<version>"` or
+`"<version>; name=<label>"`).
+
+### 5. Object-event webhooks
+
+Configure webhook notifications on object events to drive event-driven
+multi-agent pipelines without polling.
+
+```python
+from tigris_boto3_ext import set_object_notifications, clear_object_notifications
+
+set_object_notifications(
+    s3_client,
+    'pipeline-bucket',
+    webhook_url='https://my-service.example/webhook',
+    event_filter='WHERE `key` REGEXP "^results/"',
+    auth_token='my-webhook-secret',
+)
+
+clear_object_notifications(s3_client, 'pipeline-bucket')
+```
+
+> **Notes**: Workspace TTL uses the standard S3 `PutBucketLifecycleConfiguration`
+> API. Force-deletion uses the Tigris `Tigris-Force-Delete` header (also
+> exposed as `delete_bucket(s3, bucket, force=True)`). Scoped credentials use
+> the AWS-IAM-compatible Tigris IAM endpoint at `https://iam.storageapi.dev`
+> (override via the `TIGRIS_IAM_ENDPOINT` env var) — `create_workspace` /
+> `create_forks` issue a `CreateAccessKey` + bucket-scoped `CreatePolicy` +
+> `AttachUserPolicy` against the standard boto3 IAM client. Webhook
+> notifications are the only piece that uses a Tigris-specific
+> `PATCH /{bucket}` JSON endpoint.
+
 ## Complete Examples
 
 ### Example 1: Backup and Restore Workflow
diff --git a/examples/agent_kit_usage.py b/examples/agent_kit_usage.py
new file mode 100644
index 0000000..396f8d1
--- /dev/null
+++ b/examples/agent_kit_usage.py
@@ -0,0 +1,196 @@
+"""Primitive-by-primitive walkthrough of the agent_kit surface.
+
+Mirrors the shape of the other ``examples/*.py`` files: each top-level
+function demonstrates one feature, the ``__main__`` runs them in order.
+
+Run against real Tigris by exporting:
+
+    export AWS_ENDPOINT_URL_S3=https://t3.storage.dev
+    export AWS_ACCESS_KEY_ID=...
+    export AWS_SECRET_ACCESS_KEY=...
+
+then::
+
+    uv run python examples/agent_kit_usage.py
+"""
+
+from __future__ import annotations
+
+import os
+import time
+import uuid
+
+import boto3
+
+from tigris_boto3_ext import (
+    Checkpoint,
+    Fork,
+    ForkSet,
+    Workspace,
+    checkpoint,
+    clear_object_notifications,
+    create_forks,
+    create_workspace,
+    restore,
+    set_object_notifications,
+    teardown_forks,
+    teardown_workspace,
+)
+
+
+def make_client():
+    """Build a boto3 S3 client from standard AWS env vars."""
+    return boto3.client(
+        "s3",
+        endpoint_url=os.environ.get("AWS_ENDPOINT_URL_S3")
+        or os.environ.get("AWS_ENDPOINT_URL")
+        or "https://t3.storage.dev",
+        aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID", "your-access-key"),
+        aws_secret_access_key=os.environ.get(
+            "AWS_SECRET_ACCESS_KEY", "your-secret-key"
+        ),
+        region_name=os.environ.get("AWS_REGION", "auto"),
+    )
+
+
+s3 = make_client()
+
+
+def _name(prefix: str) -> str:
+    """Generate a short, unique bucket name for one example run."""
+    return f"{prefix}-{uuid.uuid4().hex[:8]}"
+
+
+def example_workspace_minimal() -> None:
+    """Minimal workspace lifecycle: create, use, tear down."""
+    print("\n=== Workspace: minimal ===")
+    bucket = _name("ak-ws")
+
+    ws: Workspace = create_workspace(s3, bucket)
+    print(f"Created workspace at bucket={ws.bucket}")
+    print(f"  snapshots enabled by default; credentials={ws.credentials}")
+
+    s3.put_object(Bucket=ws.bucket, Key="hello.txt", Body=b"hi")
+    print("  wrote hello.txt to the workspace")
+
+    teardown_workspace(s3, ws)
+    print("  torn down (force-delete via Tigris-Force-Delete header)")
+
+
+def example_workspace_with_ttl_and_credentials() -> None:
+    """Workspace with TTL and a bucket-scoped Editor access key."""
+    print("\n=== Workspace: TTL + scoped credentials ===")
+    bucket = _name("ak-ws")
+
+    ws = create_workspace(
+        s3,
+        bucket,
+        ttl_days=1,
+        credentials_role="Editor",
+    )
+    print(f"Created workspace at bucket={ws.bucket}")
+    if ws.credentials:
+        print(f"  scoped access key id: {ws.credentials.access_key_id[:6]}...")
+        print(f"  policy ARN: {ws.credentials.policy_arn}")
+        print("  (these credentials only have access to this one bucket)")
+
+    teardown_workspace(s3, ws)
+    print("  torn down (revokes credentials, deletes bucket)")
+
+
+def example_forks_parallel_runs() -> None:
+    """Snapshot a base bucket then fork it N times for parallel agents."""
+    print("\n=== Forks: parallel runs ===")
+    base = _name("ak-base")
+    create_workspace(s3, base)
+    s3.put_object(Bucket=base, Key="seed.txt", Body=b"shared-seed")
+
+    forks: ForkSet = create_forks(
+        s3,
+        base,
+        count=3,
+        prefix=_name("ak-fk"),
+    )
+    print(
+        f"Snapshotted {forks.base_bucket} ({forks.snapshot_id}) → {len(forks.forks)} forks"
+    )
+
+    # Each fork is independent — divergent writes don't affect the base or peers.
+    for i, fork in enumerate(forks.forks):
+        s3.put_object(Bucket=fork.bucket, Key="result.txt", Body=f"agent-{i}".encode())
+    for i, fork in enumerate(forks.forks):
+        body = s3.get_object(Bucket=fork.bucket, Key="result.txt")["Body"].read()
+        print(f"  fork {i}: bucket={fork.bucket}  result={body.decode()!r}")
+
+    base_seed = s3.get_object(Bucket=base, Key="seed.txt")["Body"].read()
+    print(f"  base bucket untouched: seed={base_seed.decode()!r}")
+
+    teardown_forks(s3, forks)
+    teardown_workspace(s3, Workspace(bucket=base))
+
+
+def example_checkpoint_and_restore() -> None:
+    """Capture a checkpoint, mutate the bucket, then restore into a fresh fork."""
+    print("\n=== Checkpoints: capture + restore ===")
+    bucket = _name("ak-ck")
+    ws = create_workspace(s3, bucket)
+    s3.put_object(Bucket=bucket, Key="state.json", Body=b'{"step": 1}')
+
+    ck: Checkpoint = checkpoint(s3, bucket, name=f"epoch-{int(time.time())}")
+    print(f"Checkpoint snapshot_id={ck.snapshot_id}  name={ck.name}")
+
+    # Drift the live bucket forward.
+    s3.put_object(Bucket=bucket, Key="state.json", Body=b'{"step": 2}')
+    s3.put_object(Bucket=bucket, Key="state.json", Body=b'{"step": 3}')
+
+    restored_bucket = restore(s3, bucket, ck.snapshot_id)
+    print(f"Restored point-in-time fork: {restored_bucket}")
+
+    body = s3.get_object(Bucket=restored_bucket, Key="state.json")["Body"].read()
+    print(f"  restored state.json={body.decode()!r}  (expected step=1)")
+    body_now = s3.get_object(Bucket=bucket, Key="state.json")["Body"].read()
+    print(f"  live state.json={body_now.decode()!r}  (drifted to step=3)")
+
+    # Tear down both: restored fork first, then the workspace.
+    teardown_forks(
+        s3,
+        ForkSet(
+            base_bucket=bucket,
+            snapshot_id=ck.snapshot_id,
+            forks=[Fork(bucket=restored_bucket)],
+        ),
+    )
+    teardown_workspace(s3, ws)
+
+
+def example_object_notifications() -> None:
+    """Wire up + clear an object-event webhook on a bucket."""
+    print("\n=== Object notifications ===")
+    bucket = _name("ak-notif")
+    ws = create_workspace(s3, bucket)
+
+    set_object_notifications(
+        s3,
+        bucket,
+        webhook_url="https://example.com/webhook",
+        event_filter='WHERE `key` REGEXP "^results/"',
+        auth_token="placeholder-token",
+    )
+    print(f"Notifications configured on {bucket}")
+
+    clear_object_notifications(s3, bucket)
+    print("Notifications cleared")
+
+    teardown_workspace(s3, ws)
+
+
+if __name__ == "__main__":
+    print("tigris-boto3-ext — agent_kit usage examples")
+    print("=" * 50)
+    example_workspace_minimal()
+    example_workspace_with_ttl_and_credentials()
+    example_forks_parallel_runs()
+    example_checkpoint_and_restore()
+    example_object_notifications()
+    print("\n" + "=" * 50)
+    print("Done.")
diff --git a/examples/multi_agent_workflow.py b/examples/multi_agent_workflow.py
new file mode 100644
index 0000000..4198818
--- /dev/null
+++ b/examples/multi_agent_workflow.py
@@ -0,0 +1,196 @@
+"""End-to-end multi-agent workflow on top of agent_kit.
+
+This is a small but realistic shape of how an agent runtime would use
+the agent_kit primitives in production:
+
+  1. **Orchestrator** spins up a workspace, seeds it with shared input
+     data, takes a checkpoint, then forks ``N`` independent buckets —
+     one per **trial agent** — each with its own least-privilege
+     ``Editor`` access key.
+
+  2. The trial agents run *concurrently* (here via a thread pool) and
+     each only sees the credentials for its own fork. They read the
+     seed, do some work, and write their results back into their fork.
+
+  3. The orchestrator collects every trial's result by reading from the
+     fork buckets, picks a winner, and tears everything down — buckets
+     and scoped IAM keys both — using ``teardown_forks`` /
+     ``teardown_workspace`` (force-delete via Tigris-Force-Delete).
+
+  4. To demonstrate restore, the workflow then drifts the workspace
+     bucket forward, calls ``restore(...)`` with the earlier checkpoint
+     id, and asserts the restored fork holds the original state.
+
+Run against real Tigris:
+
+    export AWS_ENDPOINT_URL_S3=https://t3.storage.dev
+    export AWS_ACCESS_KEY_ID=...
+    export AWS_SECRET_ACCESS_KEY=...
+    uv run python examples/multi_agent_workflow.py
+"""
+
+from __future__ import annotations
+
+import os
+import uuid
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from dataclasses import dataclass
+from typing import Any
+
+import boto3
+
+from tigris_boto3_ext import (
+    Fork,
+    ForkSet,
+    Workspace,
+    checkpoint,
+    create_forks,
+    create_workspace,
+    delete_bucket,
+    restore,
+    teardown_forks,
+    teardown_workspace,
+)
+
+NUM_TRIALS = 4
+SEED_KEY = "input/seed.txt"
+RESULT_KEY = "output/result.txt"
+
+
+def _orchestrator_client() -> Any:
+    """The orchestrator's S3 client — uses the operator's full-power creds."""
+    return boto3.client(
+        "s3",
+        endpoint_url=os.environ.get("AWS_ENDPOINT_URL_S3")
+        or os.environ.get("AWS_ENDPOINT_URL")
+        or "https://t3.storage.dev",
+        aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID", "your-access-key"),
+        aws_secret_access_key=os.environ.get(
+            "AWS_SECRET_ACCESS_KEY", "your-secret-key"
+        ),
+        region_name=os.environ.get("AWS_REGION", "auto"),
+    )
+
+
+def _agent_client(fork: Fork) -> Any:
+    """An S3 client that ONLY has access to one fork bucket.
+
+    The trial agent never sees the orchestrator's keys; this is the whole
+    point of provisioning per-fork credentials with ``credentials_role``.
+    """
+    if fork.credentials is None:
+        msg = f"fork {fork.bucket!r} has no scoped credentials"
+        raise RuntimeError(msg)
+    return boto3.client(
+        "s3",
+        endpoint_url=os.environ.get("AWS_ENDPOINT_URL_S3")
+        or os.environ.get("AWS_ENDPOINT_URL")
+        or "https://t3.storage.dev",
+        aws_access_key_id=fork.credentials.access_key_id,
+        aws_secret_access_key=fork.credentials.secret_access_key,
+        region_name=os.environ.get("AWS_REGION", "auto"),
+    )
+
+
+@dataclass
+class TrialResult:
+    trial_id: int
+    fork_bucket: str
+    output: str
+
+
+def run_trial(trial_id: int, fork: Fork) -> TrialResult:
+    """A trial agent: read the seed from its fork, do work, write a result.
+
+    The agent uses ONLY its scoped credentials — it has no way to reach
+    the base workspace or any sibling fork.
+    """
+    client = _agent_client(fork)
+    seed = client.get_object(Bucket=fork.bucket, Key=SEED_KEY)["Body"].read().decode()
+    output = f"trial-{trial_id}: processed {seed!r} (len={len(seed)})"
+    client.put_object(Bucket=fork.bucket, Key=RESULT_KEY, Body=output.encode())
+    return TrialResult(trial_id=trial_id, fork_bucket=fork.bucket, output=output)
+
+
+def fan_out_trials(forks: ForkSet) -> list[TrialResult]:
+    """Run all trial agents concurrently and collect their results."""
+    results: list[TrialResult] = []
+    with ThreadPoolExecutor(max_workers=len(forks.forks)) as pool:
+        futures = {
+            pool.submit(run_trial, i, fork): i for i, fork in enumerate(forks.forks)
+        }
+        for future in as_completed(futures):
+            results.append(future.result())
+    results.sort(key=lambda r: r.trial_id)
+    return results
+
+
+def main() -> None:
+    s3 = _orchestrator_client()
+    run_id = uuid.uuid4().hex[:8]
+
+    # Stage 1 — orchestrator sets up the workspace and seeds shared input.
+    workspace_name = f"akwf-{run_id}"
+    print(f"[orchestrator] create_workspace({workspace_name})")
+    ws: Workspace = create_workspace(s3, workspace_name)
+    s3.put_object(Bucket=ws.bucket, Key=SEED_KEY, Body=b"hello-from-orchestrator")
+
+    forks: ForkSet | None = None
+    try:
+        # Stage 2 — capture a checkpoint of the seed state for later restore.
+        ck = checkpoint(s3, ws.bucket, name=f"seed-{run_id}")
+        print(f"[orchestrator] checkpoint snapshot_id={ck.snapshot_id}")
+
+        # Stage 3 — fork N times, one Editor-scoped access key per fork.
+        forks = create_forks(
+            s3,
+            ws.bucket,
+            count=NUM_TRIALS,
+            prefix=f"akwf-{run_id}-trial",
+            credentials_role="Editor",
+        )
+        for fork in forks.forks:
+            scoped = "yes" if fork.credentials else "no"
+            print(f"[orchestrator]   fork={fork.bucket}  scoped-creds={scoped}")
+
+        # Stage 4 — fan out the trial agents (each uses ONLY its own creds).
+        print(f"[orchestrator] running {NUM_TRIALS} trial agents in parallel...")
+        results = fan_out_trials(forks)
+
+        # Stage 5 — orchestrator (with full creds) collects every fork's result.
+        print("[orchestrator] collecting results:")
+        for r in results:
+            persisted = (
+                s3.get_object(Bucket=r.fork_bucket, Key=RESULT_KEY)["Body"]
+                .read()
+                .decode()
+            )
+            print(f"[orchestrator]   {r.trial_id}: {persisted}")
+        winner = max(results, key=lambda r: len(r.output))
+        print(f"[orchestrator] winner: trial-{winner.trial_id}")
+
+        # Stage 6 — drift the workspace and demonstrate restore.
+        s3.put_object(Bucket=ws.bucket, Key=SEED_KEY, Body=b"polluted")
+        restored = restore(s3, ws.bucket, ck.snapshot_id)
+        try:
+            seed_after = (
+                s3.get_object(Bucket=restored, Key=SEED_KEY)["Body"].read().decode()
+            )
+            print(f"[orchestrator] restored fork {restored!r} seed={seed_after!r}")
+            if seed_after != "hello-from-orchestrator":
+                msg = f"restore returned unexpected seed: {seed_after!r}"
+                raise RuntimeError(msg)
+        finally:
+            # Tear down the standalone restore fork directly.
+            delete_bucket(s3, restored, force=True)
+    finally:
+        # Stage 7 — clean up everything: forks (revoke + delete), then workspace.
+        if forks is not None:
+            print("[orchestrator] teardown_forks")
+            teardown_forks(s3, forks)
+        print("[orchestrator] teardown_workspace")
+        teardown_workspace(s3, ws)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/examples/sprite_agent.py b/examples/sprite_agent.py
new file mode 100644
index 0000000..5ddd22c
--- /dev/null
+++ b/examples/sprite_agent.py
@@ -0,0 +1,41 @@
+"""Trial agent that runs inside one sprite (one fork bucket).
+
+Reads scoped Tigris credentials, the fork bucket name, and a trial id from
+env vars (set by the orchestrator via ``sprite exec --env``). Reads a seed
+from the fork, does some trivial work, and writes a result. The agent has
+no access to the orchestrator's credentials or any sibling fork — only
+its own bucket-scoped Editor key.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+
+import boto3
+
+SEED_KEY = "input/seed.txt"
+RESULT_KEY = "output/result.txt"
+
+
+def main() -> int:
+    bucket = os.environ["FORK_BUCKET"]
+    trial_id = os.environ["TRIAL_ID"]
+
+    s3 = boto3.client(
+        "s3",
+        endpoint_url=os.environ["AWS_ENDPOINT_URL_S3"],
+        aws_access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
+        aws_secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"],
+        region_name=os.environ.get("AWS_REGION", "auto"),
+    )
+
+    seed = s3.get_object(Bucket=bucket, Key=SEED_KEY)["Body"].read().decode()
+    output = f"trial-{trial_id} on sprite {os.uname().nodename}: processed {seed!r}"
+    s3.put_object(Bucket=bucket, Key=RESULT_KEY, Body=output.encode())
+    print(output)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/examples/sprites_multi_agent_test.py b/examples/sprites_multi_agent_test.py
new file mode 100644
index 0000000..f0ade94
--- /dev/null
+++ b/examples/sprites_multi_agent_test.py
@@ -0,0 +1,230 @@
+"""Run the multi-agent agent_kit workflow on real sprites.dev sprites.
+
+Same flow as ``examples/multi_agent_workflow.py`` but each "trial agent"
+runs inside its own sprites.dev microVM instead of a thread. The
+orchestrator:
+
+  1. Builds a workspace + N forks with scoped Editor credentials via
+     ``create_workspace`` / ``create_forks``.
+  2. For each fork: spins up a sprite, installs boto3, uploads
+     ``examples/sprite_agent.py``, runs it with the fork's scoped
+     credentials passed in as env vars.
+  3. Reads each fork bucket to collect results.
+  4. Destroys every sprite and tears down the agent_kit resources.
+
+Prereqs:
+
+  - The ``sprite`` CLI installed and authed (``sprite login``).
+  - Tigris creds in env: ``AWS_ACCESS_KEY_ID``, ``AWS_SECRET_ACCESS_KEY``,
+    ``AWS_ENDPOINT_URL_S3``.
+
+Run::
+
+    uv run python examples/sprites_multi_agent_test.py
+"""
+
+from __future__ import annotations
+
+import os
+import shlex
+import subprocess
+import sys
+import uuid
+from pathlib import Path
+from typing import Any
+
+import boto3
+
+from tigris_boto3_ext import (
+    Fork,
+    ForkSet,
+    Workspace,
+    create_forks,
+    create_workspace,
+    teardown_forks,
+    teardown_workspace,
+)
+
+NUM_TRIALS = 4
+SEED_KEY = "input/seed.txt"
+RESULT_KEY = "output/result.txt"
+
+AGENT_SCRIPT = Path(__file__).parent / "sprite_agent.py"
+
+
+def _orchestrator_client() -> Any:
+    return boto3.client(
+        "s3",
+        endpoint_url=os.environ.get("AWS_ENDPOINT_URL_S3")
+        or os.environ.get("AWS_ENDPOINT_URL")
+        or "https://t3.storage.dev",
+        aws_access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
+        aws_secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"],
+        region_name=os.environ.get("AWS_REGION", "auto"),
+    )
+
+
+_REDACT_PREFIXES = (
+    "AWS_ACCESS_KEY_ID=",
+    "AWS_SECRET_ACCESS_KEY=",
+    "AWS_SESSION_TOKEN=",
+)
+
+
+def _redact(arg: str) -> str:
+    """Hide credential values when echoing the command line.
+
+    Handles both standalone ``KEY=value`` args and the
+    ``KEY=value,KEY2=value2`` form ``sprite exec --env`` expects.
+    """
+    parts = arg.split(",")
+    redacted = []
+    changed = False
+    for part in parts:
+        for prefix in _REDACT_PREFIXES:
+            if part.startswith(prefix):
+                redacted.append(f"{prefix}<redacted>")
+                changed = True
+                break
+        else:
+            redacted.append(part)
+    return ",".join(redacted) if changed else arg
+
+
+def _sprite(*args: str, capture: bool = False) -> subprocess.CompletedProcess:
+    """Invoke the ``sprite`` CLI with logging. Raises on non-zero exit."""
+    cmd = ["sprite", *args]
+    safe = " ".join(shlex.quote(_redact(a)) for a in cmd)
+    print(f"  $ {safe}")
+    return subprocess.run(  # noqa: S603
+        cmd,
+        check=True,
+        capture_output=capture,
+        text=True,
+    )
+
+
+def provision_sprite(name: str) -> None:
+    """Create a sprite and install boto3 inside it."""
+    _sprite("create", name, "--skip-console")
+    # Sprites ship with Python; install boto3 once per sprite.
+    # `--` separates sprite-cli flags from the command's flags so `bash -c`
+    # isn't reinterpreted as a sprite flag.
+    # Use Debian's packaged boto3 — pip3 in --user mode left transitive
+    # deps off Python 3.13's sys.path on the default sprite image, so apt
+    # is more reliable. Sprite exec runs as a non-root user, so sudo.
+    _sprite(
+        "exec",
+        "-s",
+        name,
+        "--",
+        "bash",
+        "-c",
+        "sudo apt-get update -qq && sudo apt-get install -y -qq python3-boto3",
+    )
+
+
+def run_agent_on_sprite(name: str, fork: Fork, trial_id: int) -> None:
+    """Upload the agent script and run it inside ``name`` with scoped creds."""
+    if fork.credentials is None:
+        msg = f"fork {fork.bucket!r} has no scoped credentials"
+        raise RuntimeError(msg)
+
+    endpoint = (
+        os.environ.get("AWS_ENDPOINT_URL_S3")
+        or os.environ.get("AWS_ENDPOINT_URL")
+        or "https://t3.storage.dev"
+    )
+
+    # ``sprite exec --env`` takes a single comma-separated KEY=value list,
+    # not repeated flags. Tigris creds don't contain commas; fail fast if
+    # that ever changes.
+    env_pairs = [
+        f"AWS_ENDPOINT_URL_S3={endpoint}",
+        f"AWS_ACCESS_KEY_ID={fork.credentials.access_key_id}",
+        f"AWS_SECRET_ACCESS_KEY={fork.credentials.secret_access_key}",
+        f"AWS_REGION={os.environ.get('AWS_REGION', 'auto')}",
+        f"FORK_BUCKET={fork.bucket}",
+        f"TRIAL_ID={trial_id}",
+    ]
+    if any("," in p for p in env_pairs):
+        msg = "env value contains a comma; sprite --env can't represent it"
+        raise RuntimeError(msg)
+
+    _sprite(
+        "exec",
+        "-s",
+        name,
+        "--file",
+        f"{AGENT_SCRIPT}:/tmp/agent.py",
+        "--env",
+        ",".join(env_pairs),
+        "--",
+        "python3",
+        "/tmp/agent.py",  # noqa: S108
+    )
+
+
+def destroy_sprite(name: str) -> None:
+    try:
+        _sprite("destroy", name, "--force")
+    except subprocess.CalledProcessError as exc:
+        print(f"  ! warning: failed to destroy sprite {name!r}: {exc}", file=sys.stderr)
+
+
+def main() -> None:
+    if not AGENT_SCRIPT.exists():
+        msg = f"missing agent script at {AGENT_SCRIPT}"
+        raise RuntimeError(msg)
+
+    s3 = _orchestrator_client()
+    run_id = uuid.uuid4().hex[:8]
+
+    workspace_name = f"akwf-{run_id}"
+    print(f"[orchestrator] create_workspace({workspace_name})")
+    ws: Workspace = create_workspace(s3, workspace_name)
+    s3.put_object(Bucket=ws.bucket, Key=SEED_KEY, Body=b"hello-from-orchestrator")
+
+    forks: ForkSet | None = None
+    sprite_names: list[str] = []
+    try:
+        forks = create_forks(
+            s3,
+            ws.bucket,
+            count=NUM_TRIALS,
+            prefix=f"akwf-{run_id}-trial",
+            credentials_role="Editor",
+        )
+        for fork in forks.forks:
+            scoped = "yes" if fork.credentials else "no"
+            print(f"[orchestrator]   fork={fork.bucket}  scoped-creds={scoped}")
+
+        for i, fork in enumerate(forks.forks):
+            sprite_name = f"akwf-{run_id}-{i}"
+            sprite_names.append(sprite_name)
+            print(f"[orchestrator] provisioning sprite {sprite_name!r}...")
+            provision_sprite(sprite_name)
+            print(f"[orchestrator] running trial {i} on {sprite_name!r}...")
+            run_agent_on_sprite(sprite_name, fork, i)
+
+        print("[orchestrator] collecting results from fork buckets:")
+        for i, fork in enumerate(forks.forks):
+            body = (
+                s3.get_object(Bucket=fork.bucket, Key=RESULT_KEY)["Body"]
+                .read()
+                .decode()
+            )
+            print(f"[orchestrator]   {i}: {body}")
+    finally:
+        for name in sprite_names:
+            print(f"[orchestrator] destroy sprite {name!r}")
+            destroy_sprite(name)
+        if forks is not None:
+            print("[orchestrator] teardown_forks")
+            teardown_forks(s3, forks)
+        print("[orchestrator] teardown_workspace")
+        teardown_workspace(s3, ws)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/pyproject.toml b/pyproject.toml
index bb16fcd..95d4cf8 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -83,6 +83,10 @@ strict_equality = true
 module = "mypy_boto3_s3.*"
 ignore_missing_imports = true
 
+[[tool.mypy.overrides]]
+module = "mypy_boto3_iam.*"
+ignore_missing_imports = true
+
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 python_files = ["test_*.py"]
diff --git a/tests/integration/test_agent_kit.py b/tests/integration/test_agent_kit.py
new file mode 100644
index 0000000..b86e136
--- /dev/null
+++ b/tests/integration/test_agent_kit.py
@@ -0,0 +1,277 @@
+"""Integration tests for agent_kit helpers (run against real Tigris)."""
+
+import time
+
+from .conftest import bucket_exists, generate_bucket_name
+
+from tigris_boto3_ext import (
+    Checkpoint,
+    Credentials,
+    Fork,
+    ForkSet,
+    Workspace,
+    checkpoint,
+    clear_object_notifications,
+    create_forks,
+    create_workspace,
+    restore,
+    set_object_notifications,
+    teardown_forks,
+    teardown_workspace,
+)
+
+
+class TestWorkspace:
+    def test_create_workspace_default_enables_snapshots(
+        self, s3_client, test_bucket_prefix, cleanup_buckets
+    ):
+        bucket = generate_bucket_name(test_bucket_prefix, "ws-default-")
+        cleanup_buckets.append(bucket)
+
+        ws = create_workspace(s3_client, bucket)
+
+        assert isinstance(ws, Workspace)
+        assert ws.bucket == bucket
+        assert bucket_exists(s3_client, bucket)
+        # Snapshots are on by default — the head_bucket header reflects it.
+        head = s3_client.head_bucket(Bucket=bucket)
+        headers = head.get("ResponseMetadata", {}).get("HTTPHeaders", {})
+        assert (
+            str(headers.get("x-tigris-enable-snapshot", "")).lower() == "true"
+        )
+
+    def test_create_workspace_disable_snapshots(
+        self, s3_client, test_bucket_prefix, cleanup_buckets
+    ):
+        bucket = generate_bucket_name(test_bucket_prefix, "ws-no-snap-")
+        cleanup_buckets.append(bucket)
+
+        ws = create_workspace(s3_client, bucket, enable_snapshots=False)
+        assert ws.bucket == bucket
+        assert bucket_exists(s3_client, bucket)
+
+    def test_create_workspace_with_ttl(
+        self, s3_client, test_bucket_prefix, cleanup_buckets
+    ):
+        bucket = generate_bucket_name(test_bucket_prefix, "ws-ttl-")
+        cleanup_buckets.append(bucket)
+
+        ws = create_workspace(s3_client, bucket, ttl_days=1)
+        assert ws.bucket == bucket
+        # Verify the lifecycle rule was actually persisted by Tigris.
+        config = s3_client.get_bucket_lifecycle_configuration(Bucket=bucket)
+        rules = config.get("Rules", [])
+        assert any(
+            rule.get("Expiration", {}).get("Days") == 1 for rule in rules
+        )
+
+    def test_teardown_workspace_force_deletes_non_empty_bucket(
+        self, s3_client, test_bucket_prefix, cleanup_buckets
+    ):
+        bucket = generate_bucket_name(test_bucket_prefix, "ws-down-")
+        cleanup_buckets.append(bucket)
+
+        ws = create_workspace(s3_client, bucket)
+        s3_client.put_object(Bucket=bucket, Key="leftover.txt", Body=b"hi")
+
+        # force=True (the default) uses Tigris's force-delete extension.
+        teardown_workspace(s3_client, ws)
+        assert not bucket_exists(s3_client, bucket)
+
+    def test_workspace_with_credentials_round_trip(
+        self, s3_client, test_bucket_prefix, cleanup_buckets
+    ):
+        bucket = generate_bucket_name(test_bucket_prefix, "ws-cred-")
+        cleanup_buckets.append(bucket)
+
+        ws = create_workspace(s3_client, bucket, credentials_role="Editor")
+        assert isinstance(ws.credentials, Credentials)
+        assert ws.credentials.access_key_id
+        assert ws.credentials.secret_access_key
+
+        teardown_workspace(s3_client, ws)
+        assert not bucket_exists(s3_client, bucket)
+
+
+class TestForks:
+    def test_create_forks_creates_n_buckets(
+        self, s3_client, test_bucket_prefix, cleanup_buckets
+    ):
+        base = generate_bucket_name(test_bucket_prefix, "forks-base-")
+        cleanup_buckets.append(base)
+
+        create_workspace(s3_client, base)
+        s3_client.put_object(Bucket=base, Key="seed.txt", Body=b"seed-data")
+
+        prefix = generate_bucket_name(test_bucket_prefix, "fork-")
+        result = create_forks(s3_client, base, 3, prefix=prefix)
+        for fork in result.forks:
+            cleanup_buckets.append(fork.bucket)
+
+        assert isinstance(result, ForkSet)
+        assert result.base_bucket == base
+        assert result.snapshot_id
+        assert len(result.forks) == 3
+        for i, fork in enumerate(result.forks):
+            assert isinstance(fork, Fork)
+            assert fork.bucket == f"{prefix}-{i}"
+            assert bucket_exists(s3_client, fork.bucket)
+            obj = s3_client.get_object(Bucket=fork.bucket, Key="seed.txt")
+            assert obj["Body"].read() == b"seed-data"
+
+    def test_default_fork_prefix_uses_snapshot_id(
+        self, s3_client, test_bucket_prefix, cleanup_buckets
+    ):
+        # Tigris snapshot ids are ~19 chars; use a short suffix so the
+        # default prefix + snapshot id + "-N" fits the 63-char bucket-name
+        # limit (the standard test prefix + uuid is already ~34 chars).
+        base = generate_bucket_name(test_bucket_prefix, "f-")
+        cleanup_buckets.append(base)
+        create_workspace(s3_client, base)
+
+        result = create_forks(s3_client, base, 1)
+        for fork in result.forks:
+            cleanup_buckets.append(fork.bucket)
+
+        assert result.forks[0].bucket == f"{base}-fork-{result.snapshot_id}-0"
+
+    def test_forks_are_isolated(
+        self, s3_client, test_bucket_prefix, cleanup_buckets
+    ):
+        base = generate_bucket_name(test_bucket_prefix, "forks-iso-")
+        cleanup_buckets.append(base)
+        create_workspace(s3_client, base)
+        s3_client.put_object(Bucket=base, Key="shared.txt", Body=b"v1")
+
+        prefix = generate_bucket_name(test_bucket_prefix, "fork-iso-")
+        result = create_forks(s3_client, base, 2, prefix=prefix)
+        for fork in result.forks:
+            cleanup_buckets.append(fork.bucket)
+
+        s3_client.put_object(
+            Bucket=result.forks[0].bucket, Key="shared.txt", Body=b"fork-0"
+        )
+        s3_client.put_object(
+            Bucket=result.forks[1].bucket, Key="shared.txt", Body=b"fork-1"
+        )
+
+        assert (
+            s3_client.get_object(
+                Bucket=result.forks[0].bucket, Key="shared.txt"
+            )["Body"].read()
+            == b"fork-0"
+        )
+        assert (
+            s3_client.get_object(
+                Bucket=result.forks[1].bucket, Key="shared.txt"
+            )["Body"].read()
+            == b"fork-1"
+        )
+        # Base bucket is untouched.
+        assert (
+            s3_client.get_object(Bucket=base, Key="shared.txt")["Body"].read()
+            == b"v1"
+        )
+
+    def test_create_forks_with_credentials(
+        self, s3_client, test_bucket_prefix, cleanup_buckets
+    ):
+        base = generate_bucket_name(test_bucket_prefix, "forks-cred-")
+        cleanup_buckets.append(base)
+        create_workspace(s3_client, base)
+
+        prefix = generate_bucket_name(test_bucket_prefix, "fork-cred-")
+        result = create_forks(
+            s3_client, base, 2, prefix=prefix, credentials_role="ReadOnly"
+        )
+        for fork in result.forks:
+            cleanup_buckets.append(fork.bucket)
+
+        for fork in result.forks:
+            assert isinstance(fork.credentials, Credentials)
+            assert fork.credentials.access_key_id
+            assert fork.credentials.secret_access_key
+        ids = [f.credentials.access_key_id for f in result.forks]
+        assert len(set(ids)) == len(ids)
+
+        teardown_forks(s3_client, result)
+
+    def test_teardown_forks_force_deletes_non_empty(
+        self, s3_client, test_bucket_prefix, cleanup_buckets
+    ):
+        base = generate_bucket_name(test_bucket_prefix, "forks-down-")
+        cleanup_buckets.append(base)
+        create_workspace(s3_client, base)
+
+        prefix = generate_bucket_name(test_bucket_prefix, "fork-down-")
+        result = create_forks(s3_client, base, 2, prefix=prefix)
+
+        for fork in result.forks:
+            s3_client.put_object(Bucket=fork.bucket, Key="x", Body=b"x")
+
+        teardown_forks(s3_client, result)
+
+        for fork in result.forks:
+            assert not bucket_exists(s3_client, fork.bucket)
+
+
+class TestCheckpoints:
+    def test_checkpoint_returns_snapshot_id(
+        self, s3_client, test_bucket_prefix, cleanup_buckets
+    ):
+        bucket = generate_bucket_name(test_bucket_prefix, "ck-")
+        cleanup_buckets.append(bucket)
+        create_workspace(s3_client, bucket)
+
+        ck = checkpoint(s3_client, bucket, name=f"epoch-{int(time.time())}")
+        assert isinstance(ck, Checkpoint)
+        assert ck.snapshot_id
+        assert ck.name is not None
+        assert ck.name.startswith("epoch-")
+        assert ck.created_at is not None
+
+    def test_restore_creates_new_fork(
+        self, s3_client, test_bucket_prefix, cleanup_buckets
+    ):
+        bucket = generate_bucket_name(test_bucket_prefix, "ck-restore-")
+        cleanup_buckets.append(bucket)
+        create_workspace(s3_client, bucket)
+        s3_client.put_object(Bucket=bucket, Key="data.txt", Body=b"original")
+
+        ck = checkpoint(s3_client, bucket)
+
+        # Mutate the original after the checkpoint.
+        s3_client.put_object(Bucket=bucket, Key="data.txt", Body=b"changed")
+
+        restored = restore(
+            s3_client,
+            bucket,
+            ck.snapshot_id,
+            fork_name=generate_bucket_name(test_bucket_prefix, "ck-restored-"),
+        )
+        cleanup_buckets.append(restored)
+
+        body = s3_client.get_object(Bucket=restored, Key="data.txt")[
+            "Body"
+        ].read()
+        assert body == b"original"
+
+
+class TestObjectNotifications:
+    def test_set_and_clear(
+        self, s3_client, test_bucket_prefix, cleanup_buckets
+    ):
+        bucket = generate_bucket_name(test_bucket_prefix, "notif-")
+        cleanup_buckets.append(bucket)
+        create_workspace(s3_client, bucket)
+
+        # The webhook URL is never invoked here — we only verify Tigris
+        # accepts the configuration.
+        set_object_notifications(
+            s3_client,
+            bucket,
+            webhook_url="https://example.com/webhook",
+            event_filter='WHERE `key` REGEXP "^results/"',
+            auth_token="test-token",  # noqa: S106
+        )
+        clear_object_notifications(s3_client, bucket)
diff --git a/tests/test_agent_kit.py b/tests/test_agent_kit.py
new file mode 100644
index 0000000..3e63136
--- /dev/null
+++ b/tests/test_agent_kit.py
@@ -0,0 +1,526 @@
+"""Unit tests for agent_kit helpers (workspaces, forks, checkpoints)."""
+
+from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from tigris_boto3_ext import (
+    Checkpoint,
+    Credentials,
+    Fork,
+    ForkSet,
+    Workspace,
+    checkpoint,
+    create_forks,
+    create_workspace,
+    restore,
+    teardown_forks,
+    teardown_workspace,
+)
+
+
+@pytest.fixture
+def s3_client():
+    return MagicMock()
+
+
+# -- create_workspace --
+
+
+class TestCreateWorkspace:
+    def test_default_enables_snapshots(self, s3_client):
+        """Snapshot support is on by default — checkpoints are the common path."""
+        with patch("tigris_boto3_ext.agent_kit.create_snapshot_bucket") as mock_snap:
+            ws = create_workspace(s3_client, "ws-1")
+        assert ws == Workspace(bucket="ws-1")
+        mock_snap.assert_called_once_with(s3_client, "ws-1")
+        s3_client.create_bucket.assert_not_called()
+        s3_client.put_bucket_lifecycle_configuration.assert_not_called()
+
+    def test_disable_snapshots_uses_plain_create_bucket(self, s3_client):
+        with patch("tigris_boto3_ext.agent_kit.create_snapshot_bucket") as mock_snap:
+            ws = create_workspace(s3_client, "ws-2", enable_snapshots=False)
+        assert ws.bucket == "ws-2"
+        mock_snap.assert_not_called()
+        s3_client.create_bucket.assert_called_once_with(Bucket="ws-2")
+
+    def test_ttl_days_sets_lifecycle_rule(self, s3_client):
+        with patch("tigris_boto3_ext.agent_kit.create_snapshot_bucket"):
+            create_workspace(s3_client, "ws-3", ttl_days=7)
+
+        s3_client.put_bucket_lifecycle_configuration.assert_called_once()
+        kwargs = s3_client.put_bucket_lifecycle_configuration.call_args.kwargs
+        assert kwargs["Bucket"] == "ws-3"
+        rules = kwargs["LifecycleConfiguration"]["Rules"]
+        assert len(rules) == 1
+        assert rules[0]["Status"] == "Enabled"
+        assert rules[0]["Expiration"] == {"Days": 7}
+        assert rules[0]["Filter"] == {"Prefix": ""}
+
+    def test_zero_ttl_raises(self, s3_client):
+        with pytest.raises(ValueError, match="ttl_days must be positive"):
+            create_workspace(s3_client, "ws-bad", ttl_days=0)
+
+    def test_negative_ttl_raises(self, s3_client):
+        with pytest.raises(ValueError, match="ttl_days must be positive"):
+            create_workspace(s3_client, "ws-bad", ttl_days=-1)
+
+
+# -- teardown_workspace --
+
+
+class TestTeardownWorkspace:
+    def test_force_delete_uses_helper(self, s3_client):
+        with patch("tigris_boto3_ext.agent_kit.delete_bucket") as mock_delete:
+            teardown_workspace(s3_client, Workspace(bucket="ws-x"))
+        mock_delete.assert_called_once_with(s3_client, "ws-x", force=True)
+
+    def test_no_force_passes_through(self, s3_client):
+        with patch("tigris_boto3_ext.agent_kit.delete_bucket") as mock_delete:
+            teardown_workspace(s3_client, Workspace(bucket="ws-x"), force=False)
+        mock_delete.assert_called_once_with(s3_client, "ws-x", force=False)
+
+
+# -- create_forks --
+
+
+class TestCreateForks:
+    def test_creates_n_forks(self, s3_client):
+        with (
+            patch("tigris_boto3_ext.agent_kit.create_snapshot") as mock_snap,
+            patch(
+                "tigris_boto3_ext.agent_kit.get_snapshot_version",
+                return_value="snap-abc",
+            ),
+            patch("tigris_boto3_ext.agent_kit.create_fork") as mock_fork,
+        ):
+            mock_snap.return_value = {}
+            result = create_forks(s3_client, "base", 3, prefix="exp")
+
+        assert result.base_bucket == "base"
+        assert result.snapshot_id == "snap-abc"
+        assert [f.bucket for f in result.forks] == ["exp-0", "exp-1", "exp-2"]
+        mock_snap.assert_called_once_with(s3_client, "base")
+        assert mock_fork.call_count == 3
+        for i, call in enumerate(mock_fork.call_args_list):
+            assert call.args[1] == f"exp-{i}"
+            assert call.args[2] == "base"
+            assert call.kwargs == {"snapshot_version": "snap-abc"}
+
+    def test_default_prefix_uses_snapshot_id(self, s3_client):
+        """Default prefix includes the snapshot id so fork names reflect lineage."""
+        with (
+            patch("tigris_boto3_ext.agent_kit.create_snapshot") as mock_snap,
+            patch(
+                "tigris_boto3_ext.agent_kit.get_snapshot_version",
+                return_value="snap-1",
+            ),
+            patch("tigris_boto3_ext.agent_kit.create_fork"),
+        ):
+            mock_snap.return_value = {}
+            result = create_forks(s3_client, "base", 1)
+
+        assert result.forks[0].bucket == "base-fork-snap-1-0"
+
+    def test_count_zero_raises(self, s3_client):
+        with pytest.raises(ValueError, match="count must be >= 1"):
+            create_forks(s3_client, "base", 0)
+
+    def test_count_negative_raises(self, s3_client):
+        with pytest.raises(ValueError, match="count must be >= 1"):
+            create_forks(s3_client, "base", -1)
+
+    def test_missing_snapshot_version_raises(self, s3_client):
+        with (
+            patch("tigris_boto3_ext.agent_kit.create_snapshot", return_value={}),
+            patch(
+                "tigris_boto3_ext.agent_kit.get_snapshot_version", return_value=None
+            ),
+        ):
+            with pytest.raises(RuntimeError, match="Could not read snapshot version"):
+                create_forks(s3_client, "base", 1)
+
+    def test_partial_failure_skips_then_continues(self, s3_client):
+        # Third fork fails; remaining attempts still proceed (no early exit).
+        with (
+            patch("tigris_boto3_ext.agent_kit.create_snapshot", return_value={}),
+            patch(
+                "tigris_boto3_ext.agent_kit.get_snapshot_version", return_value="v1"
+            ),
+            patch("tigris_boto3_ext.agent_kit.create_fork") as mock_fork,
+        ):
+            mock_fork.side_effect = [None, None, RuntimeError("boom"), None, None]
+            result = create_forks(s3_client, "base", 5, prefix="p")
+
+        # p-2 is skipped; the other four were still attempted and recorded.
+        assert [f.bucket for f in result.forks] == ["p-0", "p-1", "p-3", "p-4"]
+        assert mock_fork.call_count == 5
+
+    def test_all_fail_raises(self, s3_client):
+        with (
+            patch("tigris_boto3_ext.agent_kit.create_snapshot", return_value={}),
+            patch(
+                "tigris_boto3_ext.agent_kit.get_snapshot_version", return_value="v1"
+            ),
+            patch(
+                "tigris_boto3_ext.agent_kit.create_fork",
+                side_effect=RuntimeError("nope"),
+            ),
+        ):
+            with pytest.raises(RuntimeError, match="Failed to create any forks"):
+                create_forks(s3_client, "base", 2)
+
+
+# -- teardown_forks --
+
+
+class TestTeardownForks:
+    def test_deletes_each_fork(self, s3_client):
+        s3_client.get_paginator.return_value.paginate.return_value = iter([])
+        fs = ForkSet(
+            base_bucket="base",
+            snapshot_id="v1",
+            forks=[Fork(bucket="f0"), Fork(bucket="f1")],
+        )
+        teardown_forks(s3_client, fs)
+        deleted = [c.kwargs["Bucket"] for c in s3_client.delete_bucket.call_args_list]
+        assert deleted == ["f0", "f1"]
+
+    def test_continues_on_error(self, s3_client):
+        s3_client.get_paginator.return_value.paginate.return_value = iter([])
+        s3_client.delete_bucket.side_effect = [RuntimeError("boom"), None]
+        fs = ForkSet(
+            base_bucket="base",
+            snapshot_id="v1",
+            forks=[Fork(bucket="f0"), Fork(bucket="f1")],
+        )
+        teardown_forks(s3_client, fs)
+        assert s3_client.delete_bucket.call_count == 2
+
+
+# -- checkpoint / restore / list_checkpoints --
+
+
+class TestCheckpoint:
+    def test_returns_dataclass_with_version_and_name(self, s3_client):
+        with (
+            patch(
+                "tigris_boto3_ext.agent_kit.create_snapshot",
+                return_value={"some": "response"},
+            ) as mock_snap,
+            patch(
+                "tigris_boto3_ext.agent_kit.get_snapshot_version",
+                return_value="snap-99",
+            ),
+        ):
+            ck = checkpoint(s3_client, "training-data", name="epoch-50")
+
+        mock_snap.assert_called_once_with(
+            s3_client, "training-data", snapshot_name="epoch-50"
+        )
+        assert ck.snapshot_id == "snap-99"
+        assert ck.name == "epoch-50"
+        assert isinstance(ck.created_at, datetime)
+        assert ck.created_at.tzinfo == timezone.utc
+
+    def test_no_name_passes_none_to_helper(self, s3_client):
+        with (
+            patch("tigris_boto3_ext.agent_kit.create_snapshot") as mock_snap,
+            patch(
+                "tigris_boto3_ext.agent_kit.get_snapshot_version",
+                return_value="snap-1",
+            ),
+        ):
+            mock_snap.return_value = {}
+            ck = checkpoint(s3_client, "b")
+        mock_snap.assert_called_once_with(s3_client, "b", snapshot_name=None)
+        assert ck.name is None
+
+    def test_missing_snapshot_version_raises(self, s3_client):
+        with (
+            patch("tigris_boto3_ext.agent_kit.create_snapshot", return_value={}),
+            patch(
+                "tigris_boto3_ext.agent_kit.get_snapshot_version", return_value=None
+            ),
+        ):
+            with pytest.raises(RuntimeError, match="Could not read snapshot version"):
+                checkpoint(s3_client, "b")
+
+
+class TestRestore:
+    def test_default_fork_name_uses_snapshot_id(self, s3_client):
+        """Default fork name embeds the snapshot id, not a timestamp."""
+        with patch("tigris_boto3_ext.agent_kit.create_fork") as mock_fork:
+            new_bucket = restore(s3_client, "training", "snap-1")
+        assert new_bucket == "training-restore-snap-1"
+        mock_fork.assert_called_once()
+        assert mock_fork.call_args.args[1] == "training-restore-snap-1"
+        assert mock_fork.call_args.args[2] == "training"
+        assert mock_fork.call_args.kwargs == {"snapshot_version": "snap-1"}
+
+    def test_custom_fork_name(self, s3_client):
+        with patch("tigris_boto3_ext.agent_kit.create_fork") as mock_fork:
+            new_bucket = restore(
+                s3_client, "training", "snap-1", fork_name="retry-1"
+            )
+        assert new_bucket == "retry-1"
+        assert mock_fork.call_args.args[1] == "retry-1"
+
+
+class TestCheckpointEquality:
+    def test_equality_ignores_created_at(self):
+        """A checkpoint() result must equal the same logical entry from a listing."""
+        client_side = Checkpoint(
+            snapshot_id="snap-1",
+            name="ep-1",
+            created_at=datetime(2026, 4, 27, tzinfo=timezone.utc),
+        )
+        server_side = Checkpoint(
+            snapshot_id="snap-1",
+            name="ep-1",
+            created_at=datetime(2026, 4, 28, tzinfo=timezone.utc),
+        )
+        assert client_side == server_side
+
+
+# -- Credentials --
+
+
+def _ak(name="bucket-key"):
+    return {
+        "access_key_id": f"AKIA-{name}",
+        "secret_access_key": "supersecret",  # noqa: S106
+        "user_name": "auto",
+        "policy_arn": f"arn:policy/{name}",
+    }
+
+
+class TestCreateWorkspaceCredentials:
+    def test_no_role_no_credentials(self, s3_client):
+        with patch(
+            "tigris_boto3_ext.agent_kit.create_scoped_access_key"
+        ) as mock_create:
+            ws = create_workspace(s3_client, "ws-x")
+        mock_create.assert_not_called()
+        assert ws.credentials is None
+
+    def test_editor_role_provisions_key(self, s3_client):
+        with patch(
+            "tigris_boto3_ext.agent_kit.create_scoped_access_key",
+            return_value=_ak("ws-1-key"),
+        ) as mock_create:
+            ws = create_workspace(s3_client, "ws-1", credentials_role="Editor")
+
+        mock_create.assert_called_once_with(
+            s3_client, "ws-1-key", "ws-1", "Editor"
+        )
+        assert ws.credentials == Credentials(
+            access_key_id="AKIA-ws-1-key",
+            secret_access_key="supersecret",  # noqa: S106
+            user_name="auto",
+            policy_arn="arn:policy/ws-1-key",
+        )
+
+    def test_readonly_role_provisions_key(self, s3_client):
+        with patch(
+            "tigris_boto3_ext.agent_kit.create_scoped_access_key",
+            return_value=_ak(),
+        ) as mock_create:
+            create_workspace(s3_client, "ws-2", credentials_role="ReadOnly")
+        assert mock_create.call_args.args[3] == "ReadOnly"
+
+    def test_invalid_role_raises(self, s3_client):
+        with pytest.raises(ValueError, match="must be 'Editor' or 'ReadOnly'"):
+            create_workspace(s3_client, "ws-bad", credentials_role="Admin")
+
+    def test_credential_failure_rolls_back_bucket(self, s3_client):
+        """If credential provisioning fails, the bucket is force-deleted."""
+        with (
+            patch("tigris_boto3_ext.agent_kit.create_snapshot_bucket"),
+            patch(
+                "tigris_boto3_ext.agent_kit.create_scoped_access_key",
+                side_effect=RuntimeError("iam down"),
+            ),
+            patch("tigris_boto3_ext.agent_kit.delete_bucket") as mock_delete,
+        ):
+            with pytest.raises(RuntimeError, match="iam down"):
+                create_workspace(
+                    s3_client, "ws-rb", credentials_role="Editor"
+                )
+        mock_delete.assert_called_once_with(s3_client, "ws-rb", force=True)
+
+    def test_invalid_role_does_not_create_bucket(self, s3_client):
+        """Bad role must fail before any bucket-creating S3 call (no leak)."""
+        with patch(
+            "tigris_boto3_ext.agent_kit.create_snapshot_bucket"
+        ) as mock_snap:
+            with pytest.raises(ValueError, match="must be 'Editor' or 'ReadOnly'"):
+                create_workspace(
+                    s3_client,
+                    "ws-bad",
+                    enable_snapshots=True,
+                    credentials_role="Admin",
+                )
+        mock_snap.assert_not_called()
+        s3_client.create_bucket.assert_not_called()
+        s3_client.put_bucket_lifecycle_configuration.assert_not_called()
+
+
+class TestTeardownWorkspaceCredentials:
+    def test_revokes_then_deletes(self, s3_client):
+        s3_client.get_paginator.return_value.paginate.return_value = iter([])
+        ws = Workspace(
+            bucket="ws-1",
+            credentials=Credentials(
+                access_key_id="AKIA",
+                secret_access_key="s",  # noqa: S106
+                user_name="auto",
+                policy_arn="arn:policy/p",
+            ),
+        )
+        with patch(
+            "tigris_boto3_ext.agent_kit.delete_scoped_access_key"
+        ) as mock_delete:
+            teardown_workspace(s3_client, ws)
+        mock_delete.assert_called_once_with(
+            s3_client,
+            access_key_id="AKIA",
+            user_name="auto",
+            policy_arn="arn:policy/p",
+        )
+        s3_client.delete_bucket.assert_called_once_with(Bucket="ws-1")
+
+    def test_revoke_failure_still_deletes_bucket(self, s3_client):
+        ws = Workspace(
+            bucket="ws-1",
+            credentials=Credentials(
+                access_key_id="AKIA",
+                secret_access_key="s",  # noqa: S106
+                user_name="auto",
+                policy_arn="arn:policy/p",
+            ),
+        )
+        with (
+            patch(
+                "tigris_boto3_ext.agent_kit.delete_scoped_access_key",
+                side_effect=RuntimeError("iam down"),
+            ),
+            patch("tigris_boto3_ext.agent_kit.delete_bucket") as mock_delete_bucket,
+        ):
+            teardown_workspace(s3_client, ws)
+        mock_delete_bucket.assert_called_once_with(s3_client, "ws-1", force=True)
+
+
+class TestCreateForksCredentials:
+    def test_per_fork_credentials(self, s3_client):
+        with (
+            patch("tigris_boto3_ext.agent_kit.create_snapshot", return_value={}),
+            patch(
+                "tigris_boto3_ext.agent_kit.get_snapshot_version", return_value="v1"
+            ),
+            patch("tigris_boto3_ext.agent_kit.create_fork"),
+            patch(
+                "tigris_boto3_ext.agent_kit.create_scoped_access_key",
+                side_effect=[_ak("f-0-key"), _ak("f-1-key")],
+            ) as mock_create,
+        ):
+            result = create_forks(
+                s3_client, "base", 2, prefix="f", credentials_role="ReadOnly"
+            )
+
+        assert mock_create.call_count == 2
+        for fork in result.forks:
+            assert fork.credentials is not None
+        # First call: name="f-0-key", bucket="f-0", role="ReadOnly"
+        first = mock_create.call_args_list[0].args
+        assert first[1] == "f-0-key"
+        assert first[2] == "f-0"
+        assert first[3] == "ReadOnly"
+
+    def test_invalid_role_does_not_create_snapshot(self, s3_client):
+        """Bad role must fail before snapshot/fork creation (no leak)."""
+        with (
+            patch("tigris_boto3_ext.agent_kit.create_snapshot") as mock_snap,
+            patch("tigris_boto3_ext.agent_kit.create_fork") as mock_fork,
+        ):
+            with pytest.raises(ValueError, match="must be 'Editor' or 'ReadOnly'"):
+                create_forks(
+                    s3_client, "base", 1, prefix="f", credentials_role="Admin"
+                )
+        mock_snap.assert_not_called()
+        mock_fork.assert_not_called()
+
+    def test_credential_failure_still_appends_fork(self, s3_client):
+        """If credential provisioning fails, the fork bucket is still tracked."""
+        with (
+            patch("tigris_boto3_ext.agent_kit.create_snapshot", return_value={}),
+            patch(
+                "tigris_boto3_ext.agent_kit.get_snapshot_version", return_value="v1"
+            ),
+            patch("tigris_boto3_ext.agent_kit.create_fork"),
+            patch(
+                "tigris_boto3_ext.agent_kit.create_scoped_access_key",
+                side_effect=[
+                    _ak("f-0-key"),
+                    RuntimeError("iam down"),
+                    _ak("f-2-key"),
+                ],
+            ),
+        ):
+            result = create_forks(
+                s3_client, "base", 3, prefix="f", credentials_role="Editor"
+            )
+
+        # All three fork buckets are tracked so teardown can clean them up.
+        assert [f.bucket for f in result.forks] == ["f-0", "f-1", "f-2"]
+        # The middle fork has no credentials but is still in the list.
+        assert result.forks[0].credentials is not None
+        assert result.forks[1].credentials is None
+        assert result.forks[2].credentials is not None
+
+
+class TestTeardownForksCredentials:
+    def test_revokes_each_credential(self, s3_client):
+        fs = ForkSet(
+            base_bucket="base",
+            snapshot_id="v1",
+            forks=[
+                Fork(
+                    bucket="f0",
+                    credentials=Credentials(
+                        access_key_id="AKIA0",
+                        secret_access_key="s",  # noqa: S106
+                        user_name="auto",
+                        policy_arn="arn:0",
+                    ),
+                ),
+                Fork(
+                    bucket="f1",
+                    credentials=Credentials(
+                        access_key_id="AKIA1",
+                        secret_access_key="s",  # noqa: S106
+                        user_name="auto",
+                        policy_arn="arn:1",
+                    ),
+                ),
+            ],
+        )
+        with (
+            patch(
+                "tigris_boto3_ext.agent_kit.delete_scoped_access_key"
+            ) as mock_delete,
+            patch("tigris_boto3_ext.agent_kit.delete_bucket") as mock_delete_bucket,
+        ):
+            teardown_forks(s3_client, fs)
+        deleted_keys = [
+            c.kwargs["access_key_id"] for c in mock_delete.call_args_list
+        ]
+        assert deleted_keys == ["AKIA0", "AKIA1"]
+        # Each fork is force-deleted via the Tigris extension.
+        assert [c.args[1] for c in mock_delete_bucket.call_args_list] == [
+            "f0",
+            "f1",
+        ]
+        for call in mock_delete_bucket.call_args_list:
+            assert call.kwargs == {"force": True}
diff --git a/tests/test_helpers.py b/tests/test_helpers.py
new file mode 100644
index 0000000..97879e2
--- /dev/null
+++ b/tests/test_helpers.py
@@ -0,0 +1,43 @@
+"""Unit tests for helper functions in helpers.py."""
+
+from unittest.mock import MagicMock, patch
+
+from tigris_boto3_ext import delete_bucket
+
+
+class TestDeleteBucket:
+    def test_default_no_force(self):
+        """Without force, just calls boto3's delete_bucket."""
+        client = MagicMock()
+        delete_bucket(client, "my-bucket")
+        client.delete_bucket.assert_called_once_with(Bucket="my-bucket")
+
+    def test_force_registers_header_injector(self):
+        """With force=True, registers the Tigris-Force-Delete header injector."""
+        client = MagicMock()
+        injector = MagicMock()
+        with patch(
+            "tigris_boto3_ext.helpers.create_header_injector", return_value=injector
+        ) as mock_factory:
+            delete_bucket(client, "my-bucket", force=True)
+
+        mock_factory.assert_called_once_with(
+            client, "DeleteBucket", {"Tigris-Force-Delete": "true"}
+        )
+        injector.register.assert_called_once()
+        injector.unregister.assert_called_once()
+        client.delete_bucket.assert_called_once_with(Bucket="my-bucket")
+
+    def test_force_unregisters_on_error(self):
+        """The header injector must be unregistered even if delete_bucket raises."""
+        client = MagicMock()
+        client.delete_bucket.side_effect = RuntimeError("boom")
+        injector = MagicMock()
+        with patch(
+            "tigris_boto3_ext.helpers.create_header_injector", return_value=injector
+        ):
+            try:
+                delete_bucket(client, "my-bucket", force=True)
+            except RuntimeError:
+                pass
+        injector.unregister.assert_called_once()
diff --git a/tests/test_iam.py b/tests/test_iam.py
new file mode 100644
index 0000000..9de29b7
--- /dev/null
+++ b/tests/test_iam.py
@@ -0,0 +1,220 @@
+"""Unit tests for the IAM helpers (boto3 IAM client wrapper)."""
+
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from tigris_boto3_ext._iam import (
+    DEFAULT_IAM_ENDPOINT,
+    _build_iam_client,
+    _policy_document,
+    create_scoped_access_key,
+    delete_scoped_access_key,
+)
+
+
+@pytest.fixture
+def s3_client():
+    client = MagicMock()
+    client.meta.region_name = "auto"
+    creds = MagicMock(access_key="ak", secret_key="sk", token=None)
+    client._request_signer._credentials.get_frozen_credentials.return_value = creds
+    return client
+
+
+@pytest.fixture
+def iam_client():
+    client = MagicMock()
+    client.create_access_key.return_value = {
+        "AccessKey": {
+            "AccessKeyId": "AKIAEXAMPLE",
+            "SecretAccessKey": "supersecret",
+            "UserName": "auto",
+        }
+    }
+    client.create_policy.return_value = {
+        "Policy": {"Arn": "arn:aws:iam::tigris:policy/bucket-policy"}
+    }
+    return client
+
+
+# -- Policy document construction --
+
+
+class TestPolicyDocument:
+    def test_editor_grants_all_s3(self):
+        doc = json.loads(_policy_document("my-bucket", "Editor"))
+        stmt = doc["Statement"][0]
+        assert stmt["Action"] == "s3:*"
+        assert stmt["Resource"] == [
+            "arn:aws:s3:::my-bucket",
+            "arn:aws:s3:::my-bucket/*",
+        ]
+        assert stmt["Effect"] == "Allow"
+
+    def test_readonly_only_grants_reads(self):
+        doc = json.loads(_policy_document("my-bucket", "ReadOnly"))
+        actions = doc["Statement"][0]["Action"]
+        assert "s3:GetObject" in actions
+        assert "s3:ListBucket" in actions
+        assert "s3:PutObject" not in actions
+        assert "s3:*" not in actions
+
+    def test_invalid_role_raises(self):
+        with pytest.raises(ValueError, match="role must be"):
+            _policy_document("b", "Admin")
+
+
+# -- Build client --
+
+
+class TestBuildIamClient:
+    @patch("tigris_boto3_ext._iam.boto3.client")
+    def test_uses_default_endpoint_and_credentials(self, mock_factory, s3_client):
+        _build_iam_client(s3_client)
+        mock_factory.assert_called_once()
+        kwargs = mock_factory.call_args.kwargs
+        assert mock_factory.call_args.args[0] == "iam"
+        assert kwargs["endpoint_url"] == DEFAULT_IAM_ENDPOINT
+        assert kwargs["aws_access_key_id"] == "ak"
+        assert kwargs["aws_secret_access_key"] == "sk"
+        assert kwargs["region_name"] == "auto"
+
+    @patch.dict("os.environ", {"TIGRIS_IAM_ENDPOINT": "https://iam.test"})
+    @patch("tigris_boto3_ext._iam.boto3.client")
+    def test_endpoint_overridable_via_env(self, mock_factory, s3_client):
+        _build_iam_client(s3_client)
+        assert (
+            mock_factory.call_args.kwargs["endpoint_url"] == "https://iam.test"
+        )
+
+
+# -- create_scoped_access_key --
+
+
+class TestCreateScopedAccessKey:
+    @patch("tigris_boto3_ext._iam.boto3.client")
+    def test_full_provisioning_flow(self, mock_factory, s3_client, iam_client):
+        mock_factory.return_value = iam_client
+
+        result = create_scoped_access_key(
+            s3_client, "ws-1-key", "ws-1", "Editor"
+        )
+
+        assert result["access_key_id"] == "AKIAEXAMPLE"
+        assert result["secret_access_key"] == "supersecret"
+        # Tigris IAM uses the access key id as the user handle.
+        assert result["user_name"] == "AKIAEXAMPLE"
+        assert result["policy_arn"] == "arn:aws:iam::tigris:policy/bucket-policy"
+
+        iam_client.create_access_key.assert_called_once_with()
+        iam_client.create_policy.assert_called_once()
+        policy_kwargs = iam_client.create_policy.call_args.kwargs
+        assert policy_kwargs["PolicyName"] == "ws-1-key-policy"
+        assert "ws-1" in policy_kwargs["PolicyDocument"]
+        iam_client.attach_user_policy.assert_called_once_with(
+            UserName="AKIAEXAMPLE",
+            PolicyArn="arn:aws:iam::tigris:policy/bucket-policy",
+        )
+
+    @patch("tigris_boto3_ext._iam.boto3.client")
+    def test_rolls_back_on_attach_failure(
+        self, mock_factory, s3_client, iam_client
+    ):
+        iam_client.attach_user_policy.side_effect = RuntimeError("denied")
+        mock_factory.return_value = iam_client
+
+        with pytest.raises(RuntimeError, match="denied"):
+            create_scoped_access_key(s3_client, "k", "b", "Editor")
+
+        # Both the orphaned access key and the policy must be cleaned up.
+        iam_client.delete_access_key.assert_called_once()
+        iam_client.delete_policy.assert_called_once_with(
+            PolicyArn="arn:aws:iam::tigris:policy/bucket-policy"
+        )
+
+    @patch("tigris_boto3_ext._iam.boto3.client")
+    def test_rolls_back_when_policy_creation_fails(
+        self, mock_factory, s3_client, iam_client
+    ):
+        iam_client.create_policy.side_effect = RuntimeError("policy error")
+        mock_factory.return_value = iam_client
+
+        with pytest.raises(RuntimeError, match="policy error"):
+            create_scoped_access_key(s3_client, "k", "b", "Editor")
+
+        iam_client.delete_access_key.assert_called_once()
+        # No policy ARN was returned, so delete_policy must not run.
+        iam_client.delete_policy.assert_not_called()
+
+    @patch("tigris_boto3_ext._iam.boto3.client")
+    def test_invalid_role_raises_before_iam_calls(
+        self, mock_factory, s3_client, iam_client
+    ):
+        mock_factory.return_value = iam_client
+        with pytest.raises(ValueError, match="role must be"):
+            create_scoped_access_key(s3_client, "k", "b", "Admin")
+        iam_client.create_access_key.assert_not_called()
+
+
+# -- delete_scoped_access_key --
+
+
+class TestDeleteScopedAccessKey:
+    @patch("tigris_boto3_ext._iam.boto3.client")
+    def test_full_teardown_order(self, mock_factory, s3_client, iam_client):
+        mock_factory.return_value = iam_client
+
+        delete_scoped_access_key(
+            s3_client,
+            access_key_id="AKIA",
+            user_name="auto",
+            policy_arn="arn:policy/p",
+        )
+
+        iam_client.detach_user_policy.assert_called_once_with(
+            UserName="auto", PolicyArn="arn:policy/p"
+        )
+        iam_client.delete_policy.assert_called_once_with(PolicyArn="arn:policy/p")
+        iam_client.delete_access_key.assert_called_once_with(AccessKeyId="AKIA")
+
+    @patch("tigris_boto3_ext._iam.boto3.client")
+    def test_continues_after_detach_failure(
+        self, mock_factory, s3_client, iam_client
+    ):
+        iam_client.detach_user_policy.side_effect = RuntimeError("err")
+        mock_factory.return_value = iam_client
+
+        delete_scoped_access_key(
+            s3_client,
+            access_key_id="AKIA",
+            user_name="auto",
+            policy_arn="arn:policy/p",
+        )
+
+        iam_client.delete_policy.assert_called_once()
+        iam_client.delete_access_key.assert_called_once()
+
+    @patch("tigris_boto3_ext._iam.boto3.client")
+    def test_skips_detach_and_policy_delete_without_arn(
+        self, mock_factory, s3_client, iam_client
+    ):
+        mock_factory.return_value = iam_client
+
+        delete_scoped_access_key(
+            s3_client, access_key_id="AKIA", user_name="auto"
+        )
+
+        iam_client.detach_user_policy.assert_not_called()
+        iam_client.delete_policy.assert_not_called()
+        iam_client.delete_access_key.assert_called_once_with(AccessKeyId="AKIA")
+
+    @patch("tigris_boto3_ext._iam.boto3.client")
+    def test_no_user_name_means_no_username_kwarg(
+        self, mock_factory, s3_client, iam_client
+    ):
+        mock_factory.return_value = iam_client
+
+        delete_scoped_access_key(s3_client, access_key_id="AKIA")
+        iam_client.delete_access_key.assert_called_once_with(AccessKeyId="AKIA")
diff --git a/tests/test_object_notifications.py b/tests/test_object_notifications.py
new file mode 100644
index 0000000..abd17e0
--- /dev/null
+++ b/tests/test_object_notifications.py
@@ -0,0 +1,168 @@
+"""Unit tests for the object_notifications public module."""
+
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from tigris_boto3_ext.object_notifications import (
+    ObjectNotificationsError,
+    clear_object_notifications,
+    set_object_notifications,
+)
+
+
+@pytest.fixture
+def mock_s3_client():
+    client = MagicMock()
+    client.meta.endpoint_url = "https://t3.storage.dev"
+    client.meta.region_name = "auto"
+    client._request_signer._credentials.get_frozen_credentials.return_value = (
+        MagicMock()
+    )
+    return client
+
+
+def _ok():
+    response = MagicMock()
+    response.status = 200
+    response.data = b""
+    return response
+
+
+class TestSetObjectNotifications:
+    @patch("tigris_boto3_ext.object_notifications._pool")
+    @patch("tigris_boto3_ext.object_notifications.SigV4Auth")
+    def test_minimal(self, mock_sigv4, mock_pool, mock_s3_client):
+        mock_pool.urlopen.return_value = _ok()
+
+        set_object_notifications(
+            mock_s3_client, "b", webhook_url="https://hook.example"
+        )
+
+        args, kwargs = mock_pool.urlopen.call_args
+        assert args[0] == "PATCH"
+        assert "/b" in args[1]
+        body = kwargs["body"]
+        if isinstance(body, bytes):
+            body = body.decode()
+        assert json.loads(body) == {
+            "object_notifications": {
+                "enabled": True,
+                "web_hook": "https://hook.example",
+                "filter": "",
+            }
+        }
+        # SigV4 needs the body hash in canonical headers.
+        assert "X-Amz-Content-Sha256" in kwargs["headers"]
+
+    @patch("tigris_boto3_ext.object_notifications._pool")
+    @patch("tigris_boto3_ext.object_notifications.SigV4Auth")
+    def test_with_filter(self, _sigv4, mock_pool, mock_s3_client):
+        mock_pool.urlopen.return_value = _ok()
+        set_object_notifications(
+            mock_s3_client,
+            "b",
+            webhook_url="https://hook.example",
+            event_filter='WHERE `key` REGEXP "^results/"',
+        )
+        body = mock_pool.urlopen.call_args.kwargs["body"]
+        if isinstance(body, bytes):
+            body = body.decode()
+        assert (
+            json.loads(body)["object_notifications"]["filter"]
+            == 'WHERE `key` REGEXP "^results/"'
+        )
+
+    @patch("tigris_boto3_ext.object_notifications._pool")
+    @patch("tigris_boto3_ext.object_notifications.SigV4Auth")
+    def test_with_token_auth(self, _sigv4, mock_pool, mock_s3_client):
+        mock_pool.urlopen.return_value = _ok()
+        set_object_notifications(
+            mock_s3_client, "b", webhook_url="https://h", auth_token="t"
+        )
+        body = mock_pool.urlopen.call_args.kwargs["body"]
+        if isinstance(body, bytes):
+            body = body.decode()
+        assert json.loads(body)["object_notifications"]["auth"] == {"token": "t"}
+
+    @patch("tigris_boto3_ext.object_notifications._pool")
+    @patch("tigris_boto3_ext.object_notifications.SigV4Auth")
+    def test_with_basic_auth(self, _sigv4, mock_pool, mock_s3_client):
+        mock_pool.urlopen.return_value = _ok()
+        set_object_notifications(
+            mock_s3_client,
+            "b",
+            webhook_url="https://h",
+            auth_username="u",
+            auth_password="p",  # noqa: S106
+        )
+        body = mock_pool.urlopen.call_args.kwargs["body"]
+        if isinstance(body, bytes):
+            body = body.decode()
+        assert json.loads(body)["object_notifications"]["auth"] == {
+            "basic_user": "u",
+            "basic_pass": "p",
+        }
+
+    def test_empty_url_raises(self, mock_s3_client):
+        with pytest.raises(ValueError, match="webhook_url is required"):
+            set_object_notifications(mock_s3_client, "b", webhook_url="")
+
+    def test_non_http_scheme_raises(self, mock_s3_client):
+        with pytest.raises(ValueError, match="must use http or https"):
+            set_object_notifications(
+                mock_s3_client, "b", webhook_url="ftp://hook"
+            )
+
+    def test_token_with_basic_raises(self, mock_s3_client):
+        with pytest.raises(ValueError, match="cannot be combined"):
+            set_object_notifications(
+                mock_s3_client,
+                "b",
+                webhook_url="https://h",
+                auth_token="t",
+                auth_username="u",
+                auth_password="p",  # noqa: S106
+            )
+
+    def test_partial_basic_auth_raises(self, mock_s3_client):
+        with pytest.raises(ValueError, match="must be provided together"):
+            set_object_notifications(
+                mock_s3_client,
+                "b",
+                webhook_url="https://h",
+                auth_username="u",
+            )
+
+    @patch("tigris_boto3_ext.object_notifications._pool")
+    @patch("tigris_boto3_ext.object_notifications.SigV4Auth")
+    def test_400_raises(self, _sigv4, mock_pool, mock_s3_client):
+        response = MagicMock()
+        response.status = 400
+        response.data = b"<Error>bad</Error>"
+        mock_pool.urlopen.return_value = response
+        with pytest.raises(ObjectNotificationsError) as exc_info:
+            set_object_notifications(
+                mock_s3_client, "b", webhook_url="https://h"
+            )
+        assert exc_info.value.status_code == 400
+        assert "bad" in exc_info.value.body
+
+    def test_empty_bucket_raises(self, mock_s3_client):
+        with pytest.raises(ValueError, match="bucket is required"):
+            set_object_notifications(
+                mock_s3_client, "", webhook_url="https://h"
+            )
+
+
+class TestClearObjectNotifications:
+    @patch("tigris_boto3_ext.object_notifications._pool")
+    @patch("tigris_boto3_ext.object_notifications.SigV4Auth")
+    def test_clears(self, _sigv4, mock_pool, mock_s3_client):
+        mock_pool.urlopen.return_value = _ok()
+        clear_object_notifications(mock_s3_client, "b")
+        body = mock_pool.urlopen.call_args.kwargs["body"]
+        if isinstance(body, bytes):
+            body = body.decode()
+        assert json.loads(body) == {"object_notifications": {}}
diff --git a/tigris_boto3_ext/__init__.py b/tigris_boto3_ext/__init__.py
index 6a1a2bd..b401bad 100644
--- a/tigris_boto3_ext/__init__.py
+++ b/tigris_boto3_ext/__init__.py
@@ -6,6 +6,19 @@
 boto3 compatibility.
 """
 
+from .agent_kit import (
+    Checkpoint,
+    Credentials,
+    Fork,
+    ForkSet,
+    Workspace,
+    checkpoint,
+    create_forks,
+    create_workspace,
+    restore,
+    teardown_forks,
+    teardown_workspace,
+)
 from .bundle import (
     BUNDLE_COMPRESSION_GZIP,
     BUNDLE_COMPRESSION_NONE,
@@ -28,6 +41,7 @@
     create_fork,
     create_snapshot,
     create_snapshot_bucket,
+    delete_bucket,
     get_bucket_info,
     get_object_from_snapshot,
     get_snapshot_version,
@@ -37,6 +51,11 @@
     list_snapshots,
     rename_object,
 )
+from .object_notifications import (
+    ObjectNotificationsError,
+    clear_object_notifications,
+    set_object_notifications,
+)
 
 __version__ = "0.3.0"
 
@@ -67,10 +86,27 @@
     "get_snapshot_version",
     "list_snapshots",
     "create_fork",
+    "delete_bucket",
     "get_object_from_snapshot",
     "list_objects_from_snapshot",
     "head_object_from_snapshot",
     "has_snapshot_enabled",
     "get_bucket_info",
     "rename_object",
+    # Agent Kit
+    "Workspace",
+    "Fork",
+    "ForkSet",
+    "Checkpoint",
+    "Credentials",
+    "create_workspace",
+    "teardown_workspace",
+    "create_forks",
+    "teardown_forks",
+    "checkpoint",
+    "restore",
+    # Object Notifications
+    "set_object_notifications",
+    "clear_object_notifications",
+    "ObjectNotificationsError",
 ]
diff --git a/tigris_boto3_ext/_iam.py b/tigris_boto3_ext/_iam.py
new file mode 100644
index 0000000..4b4cdd2
--- /dev/null
+++ b/tigris_boto3_ext/_iam.py
@@ -0,0 +1,171 @@
+"""Internal IAM helpers for provisioning bucket-scoped Tigris access keys.
+
+Tigris IAM is AWS-IAM-compatible, so we use the standard boto3 IAM client
+pointed at ``https://iam.storageapi.dev`` (override with the
+``TIGRIS_IAM_ENDPOINT`` env var). The bucket-scoped key flow is:
+
+    1. ``create_access_key()`` — provision a fresh Tigris access key.
+    2. ``create_policy(PolicyName, PolicyDocument)`` — bucket-scoped
+       policy granting the requested role (Editor → ``s3:*``;
+       ReadOnly → GetObject + ListBucket + GetBucketLocation).
+    3. ``attach_user_policy(UserName, PolicyArn)`` — wire the policy to
+       the access key. Tigris treats the access key id itself as the
+       ``UserName`` handle here, *not* the ``UserName`` field returned
+       by ``create_access_key`` (which is informational, typically
+       ``"auto"``).
+
+Teardown reverses the sequence (detach → delete policy → delete access key).
+"""
+
+import json
+import os
+from contextlib import suppress
+from typing import TYPE_CHECKING, Any, Optional
+
+import boto3
+
+if TYPE_CHECKING:
+    from mypy_boto3_iam.client import IAMClient
+    from mypy_boto3_s3.client import S3Client
+else:
+    IAMClient = object
+    S3Client = object
+
+
+DEFAULT_IAM_ENDPOINT = "https://iam.storageapi.dev"
+
+
+def _iam_endpoint() -> str:
+    return os.environ.get("TIGRIS_IAM_ENDPOINT") or DEFAULT_IAM_ENDPOINT
+
+
+def _build_iam_client(s3_client: S3Client) -> IAMClient:
+    """Build a boto3 IAM client reusing the S3 client's credentials and region."""
+    creds = s3_client._request_signer._credentials.get_frozen_credentials()  # type: ignore[attr-defined]  # noqa: SLF001
+    return boto3.client(
+        "iam",
+        endpoint_url=_iam_endpoint(),
+        aws_access_key_id=creds.access_key,
+        aws_secret_access_key=creds.secret_key,
+        aws_session_token=creds.token,
+        region_name=s3_client.meta.region_name or "auto",
+    )
+
+
+def _policy_document(bucket: str, role: str) -> str:
+    """Build a bucket-scoped IAM policy document for the given role."""
+    if role == "Editor":
+        actions: Any = "s3:*"
+    elif role == "ReadOnly":
+        actions = [
+            "s3:GetObject",
+            "s3:ListBucket",
+            "s3:GetBucketLocation",
+        ]
+    else:
+        msg = f"role must be 'Editor' or 'ReadOnly', got {role!r}"
+        raise ValueError(msg)
+
+    return json.dumps(
+        {
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Effect": "Allow",
+                    "Action": actions,
+                    "Resource": [
+                        f"arn:aws:s3:::{bucket}",
+                        f"arn:aws:s3:::{bucket}/*",
+                    ],
+                },
+            ],
+        }
+    )
+
+
+def create_scoped_access_key(
+    s3_client: S3Client,
+    name: str,
+    bucket: str,
+    role: str,
+) -> dict[str, str]:
+    """Create a bucket-scoped Tigris access key.
+
+    Args:
+        s3_client: boto3 S3 client. Credentials and region are reused.
+        name: Logical name. Used as the policy name (suffixed with
+            ``-policy``) and any other naming hooks.
+        bucket: Bucket the access key should be scoped to.
+        role: ``"Editor"`` or ``"ReadOnly"``.
+
+    Returns:
+        Dict with ``access_key_id``, ``secret_access_key``, ``user_name``,
+        and ``policy_arn`` — the last two are needed for teardown.
+
+    Raises:
+        ValueError: If ``role`` is not a recognized value.
+        botocore.exceptions.ClientError: On IAM-side failures.
+    """
+    iam = _build_iam_client(s3_client)
+    policy_doc = _policy_document(bucket, role)
+
+    key_resp = iam.create_access_key()
+    access_key = key_resp["AccessKey"]
+    access_key_id = access_key["AccessKeyId"]
+    secret = access_key["SecretAccessKey"]
+
+    # Tigris IAM treats the access key id as the addressable "user" for
+    # AttachUserPolicy. The UserName field in create_access_key's response
+    # (typically "auto") is informational, not the handle to use here.
+    policy_arn: Optional[str] = None
+    try:
+        policy_resp = iam.create_policy(
+            PolicyName=f"{name}-policy",
+            PolicyDocument=policy_doc,
+        )
+        policy_arn = policy_resp["Policy"]["Arn"]
+        iam.attach_user_policy(UserName=access_key_id, PolicyArn=policy_arn)
+    except Exception:
+        # If policy creation/attach fails, roll back the bare access key
+        # so we don't leak unscoped credentials.
+        with suppress(Exception):
+            iam.delete_access_key(AccessKeyId=access_key_id)
+        if policy_arn is not None:
+            with suppress(Exception):
+                iam.delete_policy(PolicyArn=policy_arn)
+        raise
+
+    return {
+        "access_key_id": access_key_id,
+        "secret_access_key": secret,
+        "user_name": access_key_id,
+        "policy_arn": policy_arn,
+    }
+
+
+def delete_scoped_access_key(
+    s3_client: S3Client,
+    *,
+    access_key_id: str,
+    user_name: Optional[str] = None,
+    policy_arn: Optional[str] = None,
+) -> None:
+    """Tear down a bucket-scoped Tigris access key.
+
+    Best-effort: each step is independently suppressed so a single IAM
+    failure can't strand the rest. Order matters — detach before deleting
+    the policy and the key.
+    """
+    iam = _build_iam_client(s3_client)
+
+    # Tigris IAM uses the access key id itself as the "user" handle for
+    # AttachUserPolicy / DetachUserPolicy.
+    detach_user = user_name or access_key_id
+
+    if policy_arn:
+        with suppress(Exception):
+            iam.detach_user_policy(UserName=detach_user, PolicyArn=policy_arn)
+        with suppress(Exception):
+            iam.delete_policy(PolicyArn=policy_arn)
+    with suppress(Exception):
+        iam.delete_access_key(AccessKeyId=access_key_id)
diff --git a/tigris_boto3_ext/agent_kit.py b/tigris_boto3_ext/agent_kit.py
new file mode 100644
index 0000000..3cdd5f9
--- /dev/null
+++ b/tigris_boto3_ext/agent_kit.py
@@ -0,0 +1,386 @@
+"""Agent-storage workflow helpers: workspaces, parallel forks, and checkpoints.
+
+A workspace is a Tigris bucket dedicated to a single agent — created with
+snapshots enabled by default, optional TTL via the standard S3 lifecycle
+API, and an optional bucket-scoped IAM access key. Forks are N independent
+copy-on-write buckets created from one base-bucket snapshot. Checkpoints
+are labeled snapshots that can be restored into a fresh fork.
+
+For event-driven coordination, use :mod:`tigris_boto3_ext.object_notifications`.
+"""
+
+from contextlib import suppress
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import TYPE_CHECKING, Optional
+
+if TYPE_CHECKING:
+    from typing import Literal
+
+    from mypy_boto3_s3.client import S3Client
+
+    Role = Literal["Editor", "ReadOnly"]
+else:
+    S3Client = object
+    Role = str
+
+from ._iam import create_scoped_access_key, delete_scoped_access_key
+from .helpers import (
+    create_fork,
+    create_snapshot,
+    create_snapshot_bucket,
+    delete_bucket,
+    get_snapshot_version,
+)
+
+
+@dataclass
+class Credentials:
+    """A scoped Tigris access key for a workspace or fork.
+
+    ``user_name`` and ``policy_arn`` are bookkeeping for teardown — kept out
+    of equality and repr so the user-facing key material is what shows up.
+    """
+
+    access_key_id: str
+    secret_access_key: str
+    user_name: Optional[str] = field(default=None, compare=False, repr=False)
+    policy_arn: Optional[str] = field(default=None, compare=False, repr=False)
+
+
+@dataclass
+class Workspace:
+    """A single-agent workspace."""
+
+    bucket: str
+    credentials: Optional[Credentials] = None
+
+
+@dataclass
+class Fork:
+    """A copy-on-write fork bucket."""
+
+    bucket: str
+    credentials: Optional[Credentials] = None
+
+
+@dataclass
+class ForkSet:
+    """A set of forks created from a single base-bucket snapshot."""
+
+    base_bucket: str
+    snapshot_id: str
+    forks: list[Fork] = field(default_factory=list)
+
+
+@dataclass
+class Checkpoint:
+    """A labeled snapshot of a bucket.
+
+    ``created_at`` is sourced client-side by :func:`checkpoint`
+    (``datetime.now``), so it's excluded from equality so logically equal
+    checkpoints from different sources still match.
+    """
+
+    snapshot_id: str
+    name: Optional[str] = None
+    created_at: Optional[datetime] = field(default=None, compare=False)
+
+
+def create_workspace(
+    s3_client: S3Client,
+    bucket: str,
+    *,
+    ttl_days: Optional[int] = None,
+    enable_snapshots: bool = True,
+    credentials_role: Optional[Role] = None,
+) -> Workspace:
+    """Create a workspace for an agent.
+
+    Args:
+        s3_client: boto3 S3 client.
+        bucket: Name for the workspace.
+        ttl_days: If set, configure a lifecycle rule that expires objects
+            after this many days.
+        enable_snapshots: If True (the default), enable snapshot support so
+            checkpoints can be taken later.
+        credentials_role: If set (``"Editor"`` or ``"ReadOnly"``), provision
+            a bucket-scoped Tigris access key and attach it to
+            ``Workspace.credentials``.
+
+    Returns:
+        The created Workspace.
+
+    Usage:
+        ws = create_workspace(
+            s3, "agent-abc",
+            ttl_days=1,
+            credentials_role="Editor",
+        )
+        teardown_workspace(s3, ws)
+    """
+    if ttl_days is not None and ttl_days <= 0:
+        msg = f"ttl_days must be positive, got {ttl_days}"
+        raise ValueError(msg)
+    _validate_role(credentials_role)
+
+    if enable_snapshots:
+        create_snapshot_bucket(s3_client, bucket)
+    else:
+        s3_client.create_bucket(Bucket=bucket)
+
+    if ttl_days is not None:
+        s3_client.put_bucket_lifecycle_configuration(
+            Bucket=bucket,
+            LifecycleConfiguration={
+                "Rules": [
+                    {
+                        "ID": "workspace-ttl",
+                        "Status": "Enabled",
+                        "Filter": {"Prefix": ""},
+                        "Expiration": {"Days": ttl_days},
+                    },
+                ],
+            },
+        )
+
+    try:
+        credentials = _provision_credentials(s3_client, bucket, credentials_role)
+    except Exception:
+        # Atomic semantics: if credentials were requested and provisioning
+        # fails, roll back the bucket so the caller isn't left without a
+        # handle to clean it up.
+        with suppress(Exception):
+            delete_bucket(s3_client, bucket, force=True)
+        raise
+    return Workspace(bucket=bucket, credentials=credentials)
+
+
+def teardown_workspace(
+    s3_client: S3Client,
+    workspace: Workspace,
+    *,
+    force: bool = True,
+) -> None:
+    """Delete a workspace and revoke its scoped credentials, if any.
+
+    With ``force=True`` (default) the bucket is force-deleted via the
+    Tigris extension even if non-empty.
+    """
+    if workspace.credentials is not None:
+        with suppress(Exception):
+            delete_scoped_access_key(
+                s3_client,
+                access_key_id=workspace.credentials.access_key_id,
+                user_name=workspace.credentials.user_name,
+                policy_arn=workspace.credentials.policy_arn,
+            )
+    delete_bucket(s3_client, workspace.bucket, force=force)
+
+
+def create_forks(
+    s3_client: S3Client,
+    base_bucket: str,
+    count: int,
+    *,
+    prefix: Optional[str] = None,
+    credentials_role: Optional[Role] = None,
+) -> ForkSet:
+    """Snapshot a bucket then create ``count`` independent copy-on-write forks.
+
+    Each fork is its own bucket; agents can read and write without affecting
+    the base bucket or each other. Fork creation is instant regardless of
+    base-bucket size.
+
+    The base bucket must have snapshots enabled (see
+    :func:`tigris_boto3_ext.create_snapshot_bucket`).
+
+    Args:
+        s3_client: boto3 S3 client.
+        base_bucket: Bucket to fork from.
+        count: Number of forks to create. Must be >= 1.
+        prefix: Optional prefix for fork bucket names. Defaults to
+            ``f"{base_bucket}-fork-{snapshot_id}"``.
+        credentials_role: If set (``"Editor"`` or ``"ReadOnly"``), provision a
+            bucket-scoped Tigris access key per fork.
+
+    Returns:
+        ForkSet with the base bucket, snapshot id, and a list of forks.
+
+    Raises:
+        ValueError: If ``count`` < 1 or ``credentials_role`` is unrecognized.
+        RuntimeError: If the snapshot version could not be read or no forks
+            could be created.
+    """
+    if count < 1:
+        msg = f"count must be >= 1, got {count}"
+        raise ValueError(msg)
+    _validate_role(credentials_role)
+
+    snapshot_response = create_snapshot(s3_client, base_bucket)
+    snapshot_id = get_snapshot_version(snapshot_response)
+    if snapshot_id is None:
+        msg = f"Could not read snapshot version for base bucket {base_bucket!r}"
+        raise RuntimeError(msg)
+
+    fork_prefix = prefix or f"{base_bucket}-fork-{snapshot_id}"
+    forks: list[Fork] = []
+
+    for i in range(count):
+        fork_name = f"{fork_prefix}-{i}"
+        if not _try_create_fork(s3_client, fork_name, base_bucket, snapshot_id):
+            # Best-effort: skip this fork and try the rest so a single
+            # failure (e.g. a name collision) doesn't strand the others.
+            continue
+        # Provision credentials best-effort: the bucket already exists, so
+        # we always append the Fork even if credential provisioning fails —
+        # otherwise the caller has no handle to tear down the bucket.
+        credentials = _try_provision_credentials(s3_client, fork_name, credentials_role)
+        forks.append(Fork(bucket=fork_name, credentials=credentials))
+
+    if not forks:
+        msg = f"Failed to create any forks of {base_bucket!r}"
+        raise RuntimeError(msg)
+
+    return ForkSet(base_bucket=base_bucket, snapshot_id=snapshot_id, forks=forks)
+
+
+def teardown_forks(
+    s3_client: S3Client,
+    fork_set: ForkSet,
+    *,
+    force: bool = True,
+) -> None:
+    """Delete every fork in a ForkSet, revoking each fork's credentials.
+
+    Best-effort: per-fork errors are swallowed so a single failure doesn't
+    strand the rest. With ``force=True`` (default) each bucket is
+    force-deleted via the Tigris extension even if non-empty.
+    """
+    for fork in fork_set.forks:
+        if fork.credentials is not None:
+            with suppress(Exception):
+                delete_scoped_access_key(
+                    s3_client,
+                    access_key_id=fork.credentials.access_key_id,
+                    user_name=fork.credentials.user_name,
+                    policy_arn=fork.credentials.policy_arn,
+                )
+        with suppress(Exception):
+            delete_bucket(s3_client, fork.bucket, force=force)
+
+
+def checkpoint(
+    s3_client: S3Client,
+    bucket: str,
+    *,
+    name: Optional[str] = None,
+) -> Checkpoint:
+    """Capture a checkpoint (labeled snapshot) of a bucket.
+
+    The bucket must have snapshots enabled.
+
+    Args:
+        s3_client: boto3 S3 client.
+        bucket: Bucket to snapshot.
+        name: Optional human-readable label (e.g. ``"epoch-50"``).
+
+    Returns:
+        Checkpoint with the snapshot id usable by :func:`restore`.
+    """
+    response = create_snapshot(s3_client, bucket, snapshot_name=name)
+    snapshot_id = get_snapshot_version(response)
+    if snapshot_id is None:
+        msg = f"Could not read snapshot version for bucket {bucket!r}"
+        raise RuntimeError(msg)
+    return Checkpoint(
+        snapshot_id=snapshot_id,
+        name=name,
+        created_at=datetime.now(timezone.utc),
+    )
+
+
+def restore(
+    s3_client: S3Client,
+    bucket: str,
+    snapshot_id: str,
+    *,
+    fork_name: Optional[str] = None,
+) -> str:
+    """Restore from a checkpoint into a new fork.
+
+    The original bucket is left untouched; this creates an independent
+    copy-on-write fork at the snapshot point.
+
+    Args:
+        s3_client: boto3 S3 client.
+        bucket: Bucket the checkpoint was taken on.
+        snapshot_id: Snapshot id from :func:`checkpoint`.
+        fork_name: Optional name for the new fork bucket. Defaults to
+            ``f"{bucket}-restore-{snapshot_id}"`` so the resulting bucket
+            name reflects which point-in-time it represents.
+
+    Returns:
+        Name of the new fork bucket.
+    """
+    new_name = fork_name or f"{bucket}-restore-{snapshot_id}"
+    create_fork(s3_client, new_name, bucket, snapshot_version=snapshot_id)
+    return new_name
+
+
+def _validate_role(role: Optional[Role]) -> None:
+    """Validate ``credentials_role`` upfront so a bad value can't strand a bucket.
+
+    Called *before* any bucket-creating S3 call: otherwise the caller would
+    get an exception on a bucket they were never handed back, leaking it.
+    """
+    if role is not None and role not in ("Editor", "ReadOnly"):
+        msg = f"credentials_role must be 'Editor' or 'ReadOnly', got {role!r}"
+        raise ValueError(msg)
+
+
+def _provision_credentials(
+    s3_client: S3Client, bucket: str, role: Optional[Role]
+) -> Optional[Credentials]:
+    """Create a Tigris scoped access key for ``bucket`` if a role is requested."""
+    if role is None:
+        return None
+    key = create_scoped_access_key(s3_client, f"{bucket}-key", bucket, role)
+    return Credentials(
+        access_key_id=key["access_key_id"],
+        secret_access_key=key["secret_access_key"],
+        user_name=key.get("user_name"),
+        policy_arn=key.get("policy_arn"),
+    )
+
+
+def _try_provision_credentials(
+    s3_client: S3Client, bucket: str, role: Optional[Role]
+) -> Optional[Credentials]:
+    """Best-effort credential provisioning. Returns None on IAM failure.
+
+    Used during fork creation: the bucket already exists, so we don't want
+    a single IAM hiccup to strand the caller without a handle to the
+    bucket. The fork is still tracked (without credentials) so teardown
+    can clean it up. The role itself is validated upfront by the caller.
+    """
+    if role is None:
+        return None
+    try:
+        return _provision_credentials(s3_client, bucket, role)
+    except Exception:
+        return None
+
+
+def _try_create_fork(
+    s3_client: S3Client,
+    fork_name: str,
+    base_bucket: str,
+    snapshot_id: str,
+) -> bool:
+    """Best-effort fork creation. Returns True on success, False on failure."""
+    try:
+        create_fork(s3_client, fork_name, base_bucket, snapshot_version=snapshot_id)
+    except Exception:
+        return False
+    return True
diff --git a/tigris_boto3_ext/helpers.py b/tigris_boto3_ext/helpers.py
index 849ab90..77f8072 100644
--- a/tigris_boto3_ext/helpers.py
+++ b/tigris_boto3_ext/helpers.py
@@ -390,3 +390,40 @@ def is_snapshot_enabled_header_set(headers: dict[str, Any]) -> bool:
     Check if the snapshot enabled header is set in the response headers.
     """
     return str(headers.get("x-tigris-enable-snapshot", "")).lower() == "true"
+
+
+def delete_bucket(
+    s3_client: S3Client,
+    bucket: str,
+    *,
+    force: bool = False,
+) -> dict[str, Any]:
+    """Delete a bucket, optionally using Tigris's force-delete extension.
+
+    With ``force=True`` the bucket is removed even when non-empty by sending
+    the ``Tigris-Force-Delete: true`` header — Tigris discards the contents
+    server-side, so callers don't need to enumerate and delete objects first.
+
+    Args:
+        s3_client: boto3 S3 client instance.
+        bucket: Name of the bucket to delete.
+        force: If True, force-delete a non-empty bucket via the Tigris
+            extension. Default False (standard S3 semantics: fails on
+            non-empty buckets).
+
+    Returns:
+        Response from the underlying ``delete_bucket`` operation.
+    """
+    if not force:
+        return cast("dict[str, Any]", s3_client.delete_bucket(Bucket=bucket))
+
+    injector = create_header_injector(
+        s3_client,
+        "DeleteBucket",
+        {"Tigris-Force-Delete": "true"},
+    )
+    try:
+        injector.register()
+        return cast("dict[str, Any]", s3_client.delete_bucket(Bucket=bucket))
+    finally:
+        injector.unregister()
diff --git a/tigris_boto3_ext/object_notifications.py b/tigris_boto3_ext/object_notifications.py
new file mode 100644
index 0000000..272fbfe
--- /dev/null
+++ b/tigris_boto3_ext/object_notifications.py
@@ -0,0 +1,143 @@
+"""Configure Tigris object-event webhook notifications on a bucket.
+
+Tigris exposes object-event webhooks via a bucket-settings extension that
+isn't part of the S3 API: a SigV4-signed ``PATCH /{bucket}`` with a JSON
+body. When an object is created, deleted, or modified Tigris POSTs a
+notification to the configured webhook URL.
+"""
+
+import hashlib
+import json
+from http import HTTPStatus
+from typing import TYPE_CHECKING, Any, Optional
+
+import urllib3
+from botocore.auth import SigV4Auth
+from botocore.awsrequest import AWSRequest
+
+if TYPE_CHECKING:
+    from mypy_boto3_s3.client import S3Client
+else:
+    S3Client = object
+
+
+_pool = urllib3.PoolManager(num_pools=2, maxsize=4)
+
+
+class ObjectNotificationsError(Exception):
+    """Raised when configuring Tigris object notifications fails."""
+
+    def __init__(self, message: str, status_code: int, body: str) -> None:
+        super().__init__(message)
+        self.status_code = status_code
+        self.body = body
+
+
+def set_object_notifications(
+    s3_client: S3Client,
+    bucket: str,
+    *,
+    webhook_url: str,
+    event_filter: Optional[str] = None,
+    auth_token: Optional[str] = None,
+    auth_username: Optional[str] = None,
+    auth_password: Optional[str] = None,
+) -> None:
+    """Enable object-event webhook notifications on a bucket.
+
+    Args:
+        s3_client: boto3 S3 client; endpoint and credentials are reused.
+        bucket: Bucket to configure.
+        webhook_url: Destination URL (http or https) Tigris will POST to.
+        event_filter: Optional Tigris filter expression (e.g. one matching
+            object keys under a prefix via REGEXP).
+        auth_token: Bearer token Tigris will send. Mutually exclusive with
+            basic auth.
+        auth_username: Basic-auth username; must be paired with
+            ``auth_password``.
+        auth_password: Basic-auth password; must be paired with
+            ``auth_username``.
+
+    Raises:
+        ValueError: For invalid arguments (empty/non-http URL, mixed auth
+            modes, partial basic-auth credentials).
+        ObjectNotificationsError: On non-2xx responses from Tigris.
+    """
+    _validate_webhook_url(webhook_url)
+    if auth_token is not None and (
+        auth_username is not None or auth_password is not None
+    ):
+        msg = "auth_token cannot be combined with auth_username/auth_password"
+        raise ValueError(msg)
+    if (auth_username is None) != (auth_password is None):
+        msg = "auth_username and auth_password must be provided together"
+        raise ValueError(msg)
+
+    notification: dict[str, Any] = {
+        "enabled": True,
+        "web_hook": webhook_url,
+        "filter": event_filter or "",
+    }
+    if auth_token is not None:
+        notification["auth"] = {"token": auth_token}
+    elif auth_username is not None and auth_password is not None:
+        notification["auth"] = {
+            "basic_user": auth_username,
+            "basic_pass": auth_password,
+        }
+
+    _update_bucket_settings(s3_client, bucket, {"object_notifications": notification})
+
+
+def clear_object_notifications(s3_client: S3Client, bucket: str) -> None:
+    """Disable webhook notifications on a bucket."""
+    _update_bucket_settings(s3_client, bucket, {"object_notifications": {}})
+
+
+def _validate_webhook_url(url: str) -> None:
+    if not url:
+        msg = "webhook_url is required"
+        raise ValueError(msg)
+    if not url.startswith(("http://", "https://")):
+        msg = f"webhook_url must use http or https, got {url!r}"
+        raise ValueError(msg)
+
+
+def _update_bucket_settings(
+    s3_client: S3Client,
+    bucket: str,
+    body: dict[str, Any],
+) -> None:
+    """SigV4-sign and send a ``PATCH /{bucket}`` with a JSON body."""
+    if not bucket:
+        msg = "bucket is required"
+        raise ValueError(msg)
+
+    endpoint = s3_client.meta.endpoint_url
+    credentials = s3_client._request_signer._credentials.get_frozen_credentials()  # type: ignore[attr-defined]  # noqa: SLF001
+    region = s3_client.meta.region_name or "auto"
+
+    url = f"{endpoint.rstrip('/')}/{bucket}"
+    payload = json.dumps(body)
+
+    headers = {
+        "Content-Type": "application/json",
+        "X-Amz-Content-Sha256": hashlib.sha256(payload.encode()).hexdigest(),
+    }
+
+    request = AWSRequest(method="PATCH", url=url, data=payload, headers=headers)
+    SigV4Auth(credentials, "s3", region).add_auth(request)
+    prepared = request.prepare()
+
+    response = _pool.urlopen(
+        prepared.method,
+        prepared.url,
+        body=prepared.body,
+        headers=dict(prepared.headers),
+    )
+
+    if response.status >= HTTPStatus.BAD_REQUEST:
+        raw = response.data or b""
+        text = raw.decode("utf-8", errors="replace") if raw else ""
+        msg = f"Object-notifications request failed (HTTP {response.status}): {text}"
+        raise ObjectNotificationsError(msg, status_code=response.status, body=text)
diff --git a/uv.lock b/uv.lock
index 3d73aec..9985e0a 100644
--- a/uv.lock
+++ b/uv.lock
@@ -1294,7 +1294,7 @@ wheels = [
 
 [[package]]
 name = "tigris-boto3-ext"
-version = "0.2.0"
+version = "0.3.0"
 source = { editable = "." }
 dependencies = [
     { name = "boto3" },