From b7ac1369fb4c7a71c52baa3d96ef2a84833b820e Mon Sep 17 00:00:00 2001
From: "woltspace-jerpint[bot]"
 <268897999+woltspace-jerpint[bot]@users.noreply.github.com>
Date: Sun, 20 Sep 2026 07:44:50 -0400
Subject: [PATCH 01/10] feat: add destination-owned dig authorization kernel

---
 container/lib/digging.py | 268 +++++++++++++++++++++++++++++++++++++++
 docs/digging-v0.md       |  53 ++++++++
 test/test_digging.py     | 136 ++++++++++++++++++++
 3 files changed, 457 insertions(+)
 create mode 100644 container/lib/digging.py
 create mode 100644 docs/digging-v0.md
 create mode 100644 test/test_digging.py

diff --git a/container/lib/digging.py b/container/lib/digging.py
new file mode 100644
index 00000000..1607370e
--- /dev/null
+++ b/container/lib/digging.py
@@ -0,0 +1,268 @@
+"""Destination-owned authorization state for experimental cross-colony digs.
+
+Wire authenticates and encrypts messages.  This module deliberately does not:
+it consumes the peer identity already verified by a transport adapter and owns
+the separate Woltspace decision about whether that peer may start one guest
+session.
+
+The first proof is intentionally only an authorization kernel.  It neither
+opens a network listener nor starts SSH.  A later session adapter may consume
+an approved grant exactly once and map its target into a disposable workspace.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import os
+import secrets
+import time
+from contextlib import contextmanager
+from dataclasses import asdict, dataclass
+from pathlib import Path
+
+
+DIG_VERSION = "woltspace-dig/v0"
+MAX_LIFETIME_SECONDS = 15 * 60
+_ID_BYTES = 16
+_CAP_BYTES = 32
+_STATES = frozenset({"pending", "approved", "active", "completed", "revoked", "expired"})
+
+
+class DigError(ValueError):
+    """An invalid request or lifecycle transition."""
+
+
+@dataclass(frozen=True)
+class DigRequest:
+    version: str
+    request_id: str
+    source_colony: str
+    source_wolt: str
+    task: str
+    target: str
+    created_at: int
+    expires_at: int
+
+    @classmethod
+    def create(
+        cls,
+        *,
+        source_colony: str,
+        source_wolt: str,
+        task: str,
+        target: str,
+        now: int | None = None,
+        lifetime_seconds: int = 10 * 60,
+    ) -> "DigRequest":
+        now = int(time.time()) if now is None else int(now)
+        if not 1 <= lifetime_seconds <= MAX_LIFETIME_SECONDS:
+            raise DigError("dig request lifetime must be between 1 and 900 seconds")
+        values = {
+            "source_colony": source_colony,
+            "source_wolt": source_wolt,
+            "task": task,
+            "target": target,
+        }
+        for name, value in values.items():
+            if not isinstance(value, str) or not value.strip():
+                raise DigError(f"{name} must be a non-empty string")
+        if len(task.encode("utf-8")) > 500:
+            raise DigError("task exceeds 500 UTF-8 bytes")
+        if target.startswith("/") or ".." in Path(target).parts:
+            raise DigError("target must be a destination-relative path")
+        return cls(
+            version=DIG_VERSION,
+            request_id=secrets.token_hex(_ID_BYTES),
+            source_colony=source_colony,
+            source_wolt=source_wolt,
+            task=task,
+            target=target,
+            created_at=now,
+            expires_at=now + lifetime_seconds,
+        )
+
+    @classmethod
+    def from_dict(cls, value: dict) -> "DigRequest":
+        if not isinstance(value, dict) or set(value) != set(cls.__dataclass_fields__):
+            raise DigError("dig request has unknown or missing fields")
+        request = cls(**value)
+        if request.version != DIG_VERSION:
+            raise DigError("unsupported dig request version")
+        for name in ("request_id", "source_colony", "source_wolt", "task", "target"):
+            item = getattr(request, name)
+            if not isinstance(item, str) or not item.strip():
+                raise DigError(f"{name} must be a non-empty string")
+        if len(request.request_id) != _ID_BYTES * 2 or any(
+            c not in "0123456789abcdef" for c in request.request_id
+        ):
+            raise DigError("invalid dig request id")
+        if len(request.task.encode("utf-8")) > 500:
+            raise DigError("task exceeds 500 UTF-8 bytes")
+        if request.target.startswith("/") or ".." in Path(request.target).parts:
+            raise DigError("target must be a destination-relative path")
+        if type(request.created_at) is not int or type(request.expires_at) is not int:
+            raise DigError("dig request times must be integers")
+        if request.expires_at <= request.created_at:
+            raise DigError("dig request expiry must follow creation")
+        if request.expires_at - request.created_at > MAX_LIFETIME_SECONDS:
+            raise DigError("dig request lifetime exceeds 900 seconds")
+        return request
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+
+class DigStore:
+    """Private destination-side store for pending requests and one-use grants."""
+
+    def __init__(self, root: str | Path):
+        self.root = Path(root)
+        self.root.mkdir(parents=True, exist_ok=True, mode=0o700)
+        os.chmod(self.root, 0o700)
+
+    def _path(self, request_id: str) -> Path:
+        if len(request_id) != _ID_BYTES * 2 or any(c not in "0123456789abcdef" for c in request_id):
+            raise DigError("invalid dig request id")
+        return self.root / f"{request_id}.json"
+
+    @contextmanager
+    def _locked(self, request_id: str):
+        lock_path = self.root / f".{request_id}.lock"
+        fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600)
+        try:
+            import fcntl
+
+            fcntl.flock(fd, fcntl.LOCK_EX)
+            yield self._path(request_id)
+        finally:
+            os.close(fd)
+
+    @staticmethod
+    def _cap_digest(capability: str) -> str:
+        return hashlib.sha256(capability.encode("ascii")).hexdigest()
+
+    def _write_new(self, path: Path, record: dict) -> None:
+        payload = (json.dumps(record, sort_keys=True, separators=(",", ":")) + "\n").encode()
+        fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        with os.fdopen(fd, "wb") as handle:
+            handle.write(payload)
+
+    def _replace(self, path: Path, record: dict) -> None:
+        temp = path.with_suffix(f".{secrets.token_hex(8)}.tmp")
+        payload = (json.dumps(record, sort_keys=True, separators=(",", ":")) + "\n").encode()
+        fd = os.open(temp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        try:
+            with os.fdopen(fd, "wb") as handle:
+                handle.write(payload)
+                handle.flush()
+                os.fsync(handle.fileno())
+            os.replace(temp, path)
+        finally:
+            try:
+                temp.unlink()
+            except FileNotFoundError:
+                pass
+
+    def receive(self, request: DigRequest, *, verified_peer: str, now: int | None = None) -> dict:
+        now = int(time.time()) if now is None else int(now)
+        if verified_peer != request.source_colony:
+            raise DigError("Wire peer does not match dig request source")
+        if now >= request.expires_at:
+            raise DigError("dig request is expired")
+        record = {"request": request.to_dict(), "state": "pending", "received_at": now}
+        self._write_new(self._path(request.request_id), record)
+        return self.status(request.request_id, now=now)
+
+    def approve(self, request_id: str, *, now: int | None = None, lifetime_seconds: int = 5 * 60) -> str:
+        now = int(time.time()) if now is None else int(now)
+        if not 1 <= lifetime_seconds <= MAX_LIFETIME_SECONDS:
+            raise DigError("dig grant lifetime must be between 1 and 900 seconds")
+        with self._locked(request_id) as path:
+            record = self._load(path)
+            self._expire(record, now)
+            if record["state"] != "pending":
+                raise DigError(f"cannot approve dig in state {record['state']}")
+            capability = secrets.token_urlsafe(_CAP_BYTES)
+            record.update(
+                state="approved",
+                approved_at=now,
+                grant_expires_at=min(now + lifetime_seconds, record["request"]["expires_at"]),
+                capability_sha256=self._cap_digest(capability),
+            )
+            self._replace(path, record)
+            return capability
+
+    def consume(self, request_id: str, capability: str, *, verified_peer: str, now: int | None = None) -> dict:
+        now = int(time.time()) if now is None else int(now)
+        with self._locked(request_id) as path:
+            record = self._load(path)
+            self._expire(record, now)
+            if record["state"] != "approved":
+                if record["state"] == "active":
+                    raise DigError("dig grant was already consumed")
+                raise DigError(f"cannot consume dig in state {record['state']}")
+            if verified_peer != record["request"]["source_colony"]:
+                raise DigError("Wire peer does not match approved source")
+            supplied = self._cap_digest(capability)
+            if not hmac.compare_digest(supplied, record["capability_sha256"]):
+                raise DigError("invalid dig capability")
+            record.update(state="active", started_at=now)
+            record.pop("capability_sha256", None)
+            self._replace(path, record)
+            return self._public(record)
+
+    def finish(self, request_id: str, *, now: int | None = None) -> dict:
+        now = int(time.time()) if now is None else int(now)
+        with self._locked(request_id) as path:
+            record = self._load(path)
+            if record["state"] != "active":
+                raise DigError(f"cannot finish dig in state {record['state']}")
+            record.update(state="completed", finished_at=now)
+            self._replace(path, record)
+            return self._public(record)
+
+    def revoke(self, request_id: str, *, now: int | None = None) -> dict:
+        now = int(time.time()) if now is None else int(now)
+        with self._locked(request_id) as path:
+            record = self._load(path)
+            if record["state"] in {"completed", "revoked", "expired"}:
+                raise DigError(f"cannot revoke dig in state {record['state']}")
+            record.update(state="revoked", revoked_at=now)
+            record.pop("capability_sha256", None)
+            self._replace(path, record)
+            return self._public(record)
+
+    def status(self, request_id: str, *, now: int | None = None) -> dict:
+        now = int(time.time()) if now is None else int(now)
+        with self._locked(request_id) as path:
+            record = self._load(path)
+            if self._expire(record, now):
+                self._replace(path, record)
+            return self._public(record)
+
+    @staticmethod
+    def _public(record: dict) -> dict:
+        public = dict(record)
+        public.pop("capability_sha256", None)
+        return public
+
+    @staticmethod
+    def _load(path: Path) -> dict:
+        try:
+            record = json.loads(path.read_text())
+        except FileNotFoundError as exc:
+            raise DigError("unknown dig request") from exc
+        if record.get("state") not in _STATES:
+            raise DigError("invalid stored dig state")
+        return record
+
+    @staticmethod
+    def _expire(record: dict, now: int) -> bool:
+        deadline = record.get("grant_expires_at", record["request"]["expires_at"])
+        if record["state"] in {"pending", "approved"} and now >= deadline:
+            record.update(state="expired", expired_at=now)
+            record.pop("capability_sha256", None)
+            return True
+        return False
diff --git a/docs/digging-v0.md b/docs/digging-v0.md
new file mode 100644
index 00000000..9f8d4e9e
--- /dev/null
+++ b/docs/digging-v0.md
@@ -0,0 +1,53 @@
+# Wolt digging v0: local experimental boundary
+
+Digging is a Woltspace feature that lets a wolt request a temporary guest work
+session in another colony. Woltspace Wire is only the authenticated, encrypted
+transport. Pairing colonies does not grant remote execution.
+
+This first slice freezes the destination-owned authorization lifecycle. It does
+not open SSH, expose a listener, start a remote agent, or protect production
+secrets.
+
+## Human experience
+
+1. Alice asks to dig to Bobeaver Colony with a short task description and a
+   destination-relative disposable workspace.
+2. Wire delivers the signed and encrypted request from Alice's pinned colony
+   identity.
+3. Bobeaver's human sees the source colony, source wolt, task, target, and short
+   expiry. They explicitly allow or reject this one dig.
+4. Allow creates a random, short-lived, one-use grant bound to the verified
+   source colony and exact request. The grant returns over Wire.
+5. Alice redeems it over the same pinned Wire relationship. Only then may
+   Woltspace create a constrained guest session.
+6. Completion, expiry, or revocation closes the dig. Results and an inert audit
+   summary may return over Wire.
+
+## Security invariants
+
+- A Wire peer is a messenger, not a local authority.
+- Pairing never implies permission to dig.
+- Approval is local, explicit, exact-request, short-lived, and one-use.
+- The request body cannot choose its authenticated source identity.
+- Targets are destination-relative and later resolve only inside a newly
+  created disposable workspace.
+- No host path, standing account, SSH private key, relay read capability, or
+  permanent shell credential crosses colonies.
+- The destination can revoke before or during a dig. Active-session termination
+  is a required integration gate, not yet implemented by the kernel.
+- The guest session receives a purpose-built policy and cannot inherit the
+  destination wolt's normal Auto grant.
+- Logs and status omit bearer capabilities and message bodies by default.
+
+## Current implementation
+
+`container/lib/digging.py` provides strict request parsing and a private durable
+state machine: `pending -> approved -> active -> completed`, with `revoked` and
+`expired` terminal paths. Capabilities are stored only as SHA-256 digests and
+removed on redemption. `verified_peer` is an input from the future Wire adapter,
+not a claim accepted from the request body.
+
+This is not yet a usable remote-access feature. The next gate is a disposable
+two-colony runner with a fake/in-memory Wire adapter. Only after lifecycle,
+termination, audit redaction, and hostile-request tests pass should an actual
+private transport be considered.
diff --git a/test/test_digging.py b/test/test_digging.py
new file mode 100644
index 00000000..be61f444
--- /dev/null
+++ b/test/test_digging.py
@@ -0,0 +1,136 @@
+import json
+import sys
+import threading
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "container" / "lib"))
+
+from digging import DIG_VERSION, DigError, DigRequest, DigStore
+
+
+def request(now=100):
+    return DigRequest.create(
+        source_colony="alice-root-key",
+        source_wolt="n00b",
+        task="Create a hello file in the disposable guest workspace",
+        target="guest-work",
+        now=now,
+        lifetime_seconds=600,
+    )
+
+
+def test_owned_colony_happy_path_is_explicit_and_one_use(tmp_path):
+    store = DigStore(tmp_path / "digs")
+    req = request()
+    pending = store.receive(req, verified_peer="alice-root-key", now=101)
+    assert pending["state"] == "pending"
+
+    capability = store.approve(req.request_id, now=102, lifetime_seconds=120)
+    assert capability not in json.dumps(store.status(req.request_id, now=103))
+    active = store.consume(
+        req.request_id, capability, verified_peer="alice-root-key", now=104
+    )
+    assert active["state"] == "active"
+    with pytest.raises(DigError, match="already consumed"):
+        store.consume(req.request_id, capability, verified_peer="alice-root-key", now=105)
+
+    assert store.finish(req.request_id, now=106)["state"] == "completed"
+
+
+def test_wire_pairing_does_not_implicitly_authorize_a_dig(tmp_path):
+    store = DigStore(tmp_path / "digs")
+    req = request()
+    store.receive(req, verified_peer="alice-root-key", now=101)
+    with pytest.raises(DigError, match="state pending"):
+        store.consume(req.request_id, "anything", verified_peer="alice-root-key", now=102)
+
+
+def test_transport_peer_must_match_claimed_source(tmp_path):
+    store = DigStore(tmp_path / "digs")
+    with pytest.raises(DigError, match="does not match"):
+        store.receive(request(), verified_peer="mallory-root-key", now=101)
+
+
+def test_grant_is_bound_to_source_peer(tmp_path):
+    store = DigStore(tmp_path / "digs")
+    req = request()
+    store.receive(req, verified_peer="alice-root-key", now=101)
+    capability = store.approve(req.request_id, now=102)
+    with pytest.raises(DigError, match="does not match"):
+        store.consume(req.request_id, capability, verified_peer="mallory-root-key", now=103)
+
+
+def test_expired_or_revoked_grants_cannot_be_consumed(tmp_path):
+    store = DigStore(tmp_path / "digs")
+    expired = request()
+    store.receive(expired, verified_peer="alice-root-key", now=101)
+    cap = store.approve(expired.request_id, now=102, lifetime_seconds=2)
+    with pytest.raises(DigError, match="state expired"):
+        store.consume(expired.request_id, cap, verified_peer="alice-root-key", now=104)
+
+    revoked = request(now=200)
+    store.receive(revoked, verified_peer="alice-root-key", now=201)
+    cap = store.approve(revoked.request_id, now=202)
+    assert store.revoke(revoked.request_id, now=203)["state"] == "revoked"
+    with pytest.raises(DigError, match="state revoked"):
+        store.consume(revoked.request_id, cap, verified_peer="alice-root-key", now=204)
+
+
+def test_request_schema_and_target_are_strict():
+    req = request()
+    payload = req.to_dict()
+    payload["surprise"] = True
+    with pytest.raises(DigError, match="unknown or missing"):
+        DigRequest.from_dict(payload)
+    with pytest.raises(DigError, match="destination-relative"):
+        DigRequest.create(
+            source_colony="alice",
+            source_wolt="n00b",
+            task="nope",
+            target="../host",
+            now=1,
+        )
+    assert req.version == DIG_VERSION
+
+
+def test_store_is_private_and_never_persists_raw_capability(tmp_path):
+    store = DigStore(tmp_path / "digs")
+    req = request()
+    store.receive(req, verified_peer="alice-root-key", now=101)
+    capability = store.approve(req.request_id, now=102)
+    path = tmp_path / "digs" / f"{req.request_id}.json"
+    assert (tmp_path / "digs").stat().st_mode & 0o777 == 0o700
+    assert path.stat().st_mode & 0o777 == 0o600
+    assert capability not in path.read_text()
+
+
+def test_only_one_concurrent_redeemer_wins(tmp_path):
+    store = DigStore(tmp_path / "digs")
+    req = request()
+    store.receive(req, verified_peer="alice-root-key", now=101)
+    capability = store.approve(req.request_id, now=102)
+    barrier = threading.Barrier(12)
+    outcomes = []
+
+    def redeem():
+        barrier.wait()
+        try:
+            store.consume(
+                req.request_id,
+                capability,
+                verified_peer="alice-root-key",
+                now=103,
+            )
+            outcomes.append("won")
+        except DigError:
+            outcomes.append("lost")
+
+    workers = [threading.Thread(target=redeem) for _ in range(12)]
+    for worker in workers:
+        worker.start()
+    for worker in workers:
+        worker.join()
+    assert outcomes.count("won") == 1
+    assert outcomes.count("lost") == 11

From c82013be326c4b9927c6beccea32ecbea9a304c3 Mon Sep 17 00:00:00 2001
From: "woltspace-jerpint[bot]"
 <268897999+woltspace-jerpint[bot]@users.noreply.github.com>
Date: Sun, 20 Sep 2026 07:55:32 -0400
Subject: [PATCH 02/10] docs: define digging ownership and callback model

---
 docs/digging-v0.md | 37 +++++++++++++++++++++++++++++++++++++
 1 file changed, 37 insertions(+)

diff --git a/docs/digging-v0.md b/docs/digging-v0.md
index 9f8d4e9e..5e584ee4 100644
--- a/docs/digging-v0.md
+++ b/docs/digging-v0.md
@@ -4,6 +4,35 @@ Digging is a Woltspace feature that lets a wolt request a temporary guest work
 session in another colony. Woltspace Wire is only the authenticated, encrypted
 transport. Pairing colonies does not grant remote execution.
 
+## Beginner model
+
+- **Home** is the one colony where the wolt exists and keeps its identity,
+  memory, configuration, and long-lived session history.
+- **Wire** lets paired colonies identify and call one another. It carries dig
+  requests, grants, callbacks, and results, but grants no execution authority.
+- **Dig** is an explicitly approved temporary visit into a destination. SSH may
+  carry that visit, but SSH is an implementation detail rather than the product
+  concept.
+- **IWCL** lets the temporary visiting session explain and coordinate with the
+  destination's resident wolts while it is there.
+
+Digging is not teleportation, installation, migration, or cloning. A visiting
+wolt does not become a resident of the destination colony.
+
+## User story
+
+A wolt proposes setting up another colony in a particular way. Its human asks
+the destination to allow a dig. After the destination human reviews and allows
+the exact visit, Woltspace creates a bounded guest session on the remote server
+under the visiting wolt's home identity. The visitor performs only the approved
+work and uses local IWCL to explain the resulting setup to the resident/main
+wolt. It returns its result home and the guest session is destroyed.
+
+The resident wolt or human may later call the visitor's home colony over Wire.
+The original wolt can answer remotely or request a fresh dig. Periodic check-ins
+are scheduled Wire callbacks/status exchanges or newly approved short visits,
+not a forgotten permanent shell or a dormant copy of the visiting wolt.
+
 This first slice freezes the destination-owned authorization lifecycle. It does
 not open SSH, expose a listener, start a remote agent, or protect production
 secrets.
@@ -23,6 +52,11 @@ secrets.
 6. Completion, expiry, or revocation closes the dig. Results and an inert audit
    summary may return over Wire.
 
+The same lifecycle may begin without Wire: an owner can manually grant a wolt a
+temporary visit to a server they control, with SSH or another Woltspace adapter
+providing transport. Wire is the preferred paired-colony convenience path, not
+a prerequisite for the general digging concept.
+
 ## Security invariants
 
 - A Wire peer is a messenger, not a local authority.
@@ -33,6 +67,9 @@ secrets.
   created disposable workspace.
 - No host path, standing account, SSH private key, relay read capability, or
   permanent shell credential crosses colonies.
+- The home colony remains authoritative for the visitor's identity and memory.
+  The destination stores only bounded visit/audit records and never materializes
+  a second resident wolt.
 - The destination can revoke before or during a dig. Active-session termination
   is a required integration gate, not yet implemented by the kernel.
 - The guest session receives a purpose-built policy and cannot inherit the

From c01d60bb0c5062fcd4962efa33626ff0d4accc03 Mon Sep 17 00:00:00 2001
From: "woltspace-jerpint[bot]"
 <268897999+woltspace-jerpint[bot]@users.noreply.github.com>
Date: Sun, 20 Sep 2026 08:00:48 -0400
Subject: [PATCH 03/10] docs: simplify digging v0 to owner-provisioned ssh

---
 docs/digging-v0.md | 40 +++++++++++++++++++++++++++++++---------
 1 file changed, 31 insertions(+), 9 deletions(-)

diff --git a/docs/digging-v0.md b/docs/digging-v0.md
index 5e584ee4..8821c09d 100644
--- a/docs/digging-v0.md
+++ b/docs/digging-v0.md
@@ -1,4 +1,4 @@
-# Wolt digging v0: local experimental boundary
+# Wolt digging: product model and experimental roadmap
 
 Digging is a Woltspace feature that lets a wolt request a temporary guest work
 session in another colony. Woltspace Wire is only the authenticated, encrypted
@@ -33,11 +33,34 @@ The original wolt can answer remotely or request a fresh dig. Periodic check-ins
 are scheduled Wire callbacks/status exchanges or newly approved short visits,
 not a forgotten permanent shell or a dormant copy of the visiting wolt.
 
-This first slice freezes the destination-owned authorization lifecycle. It does
-not open SSH, expose a listener, start a remote agent, or protect production
-secrets.
+## Runnable v0: owner-provisioned SSH
 
-## Human experience
+The owner configures ordinary SSH access using their existing OS user,
+`~/.ssh/config`, agent, and keys. Woltspace records which wolt may dig to the
+exact SSH host/user (and optionally a working directory), invokes normal SSH,
+and gives the wolt the same remote shell and Woltspace CLI access that owner has
+already authorized. The wolt can inspect or set up the machine, disconnect, and
+report what changed.
+
+V0 does not create a guest Unix account, transport private SSH keys over Wire,
+require a remote Woltspace agent, or provide IWCL across the colony boundary.
+The SSH user's existing authority is the real authority; the Woltspace record
+is an understandable local consent and audit boundary, not a sandbox.
+
+V0 must still pin or verify the SSH host key, bind consent to the exact host and
+user, handle remote PATH and TTY behavior, distinguish home-wolt context from
+remote resident files, report interrupted setup honestly, and provide a clear
+way to remove both the Woltspace grant and underlying SSH access.
+
+## V1: paired-colony visit
+
+Wire carries the request, destination approval, callback, and result. The
+destination-owned authorization kernel in this branch belongs here. A later V1
+may let the visiting session coordinate with the resident/main wolt over local
+IWCL. That requires an explicit identity/context bridge so the visitor never
+adopts the resident wolt's memory or becomes a remote clone.
+
+## V1 human experience
 
 1. Alice asks to dig to Bobeaver Colony with a short task description and a
    destination-relative disposable workspace.
@@ -84,7 +107,6 @@ state machine: `pending -> approved -> active -> completed`, with `revoked` and
 removed on redemption. `verified_peer` is an input from the future Wire adapter,
 not a claim accepted from the request body.
 
-This is not yet a usable remote-access feature. The next gate is a disposable
-two-colony runner with a fake/in-memory Wire adapter. Only after lifecycle,
-termination, audit redaction, and hostile-request tests pass should an actual
-private transport be considered.
+This is not yet a usable remote-access feature. The next implementation gate is
+the simpler SSH v0. The Wire state machine remains isolated V1 groundwork until
+the SSH setup/access experience and its audit boundary are proven.

From 70c20c1036be6fd16c5815cb9d135e416e03ab20 Mon Sep 17 00:00:00 2001
From: "woltspace-jerpint[bot]"
 <268897999+woltspace-jerpint[bot]@users.noreply.github.com>
Date: Sun, 20 Sep 2026 08:20:03 -0400
Subject: [PATCH 04/10] docs: reduce digging v0 to ssh bootstrap handoff

---
 docs/digging-v0.md | 32 ++++++++++++++++----------------
 1 file changed, 16 insertions(+), 16 deletions(-)

diff --git a/docs/digging-v0.md b/docs/digging-v0.md
index 8821c09d..fd93136d 100644
--- a/docs/digging-v0.md
+++ b/docs/digging-v0.md
@@ -19,19 +19,18 @@ transport. Pairing colonies does not grant remote execution.
 Digging is not teleportation, installation, migration, or cloning. A visiting
 wolt does not become a resident of the destination colony.
 
-## User story
+## V0 user story
 
-A wolt proposes setting up another colony in a particular way. Its human asks
-the destination to allow a dig. After the destination human reviews and allows
-the exact visit, Woltspace creates a bounded guest session on the remote server
-under the visiting wolt's home identity. The visitor performs only the approved
-work and uses local IWCL to explain the resulting setup to the resident/main
-wolt. It returns its result home and the guest session is destroyed.
+A human lets a wolt dig into a machine they control. The wolt connects through
+ordinary SSH, installs or configures what the machine needs, and leaves durable
+local handoff files for the new colony's wolts to find when they wake up. It
+disconnects and reports home what it changed and verified.
 
-The resident wolt or human may later call the visitor's home colony over Wire.
-The original wolt can answer remotely or request a fresh dig. Periodic check-ins
-are scheduled Wire callbacks/status exchanges or newly approved short visits,
-not a forgotten permanent shell or a dormant copy of the visiting wolt.
+The handoff may contain a bootstrap note, setup manifest, decisions, next steps,
+and verification results. It belongs in an intentionally designated shared
+bootstrap location, never another wolt's private memory.
+
+In short: **connect -> set up -> leave a local handoff -> return**.
 
 ## Runnable v0: owner-provisioned SSH
 
@@ -52,13 +51,14 @@ user, handle remote PATH and TTY behavior, distinguish home-wolt context from
 remote resident files, report interrupted setup honestly, and provide a clear
 way to remove both the Woltspace grant and underlying SSH access.
 
-## V1: paired-colony visit
+## Later: paired-colony visits
 
 Wire carries the request, destination approval, callback, and result. The
-destination-owned authorization kernel in this branch belongs here. A later V1
-may let the visiting session coordinate with the resident/main wolt over local
-IWCL. That requires an explicit identity/context bridge so the visitor never
-adopts the resident wolt's memory or becomes a remote clone.
+destination-owned authorization kernel in this branch belongs here.
+
+Cross-tunnel IWCL, a visiting-wolt identity bridge, callbacks, and ongoing
+collaboration are explicitly out of v0. Real digging experience should tell us
+whether they are needed and what identity and consent model they require.
 
 ## V1 human experience
 

From 23edbe904cf723b764472d74c5d4206084f5f276 Mon Sep 17 00:00:00 2001
From: "woltspace-jerpint[bot]"
 <268897999+woltspace-jerpint[bot]@users.noreply.github.com>
Date: Sun, 20 Sep 2026 08:24:33 -0400
Subject: [PATCH 05/10] feat: add owner-approved ssh digging v0

---
 docs/digging-v0.md   |  27 +++++
 pyproject.toml       |   2 +
 src/woltspace/cli.py | 126 ++++++++++++++++++++++-
 src/woltspace/dig.py | 239 +++++++++++++++++++++++++++++++++++++++++++
 test/test_dig_cli.py | 141 +++++++++++++++++++++++++
 5 files changed, 534 insertions(+), 1 deletion(-)
 create mode 100644 src/woltspace/dig.py
 create mode 100644 test/test_dig_cli.py

diff --git a/docs/digging-v0.md b/docs/digging-v0.md
index fd93136d..e24c874c 100644
--- a/docs/digging-v0.md
+++ b/docs/digging-v0.md
@@ -51,6 +51,33 @@ user, handle remote PATH and TTY behavior, distinguish home-wolt context from
 remote resident files, report interrupted setup honestly, and provide a clear
 way to remove both the Woltspace grant and underlying SSH access.
 
+### CLI
+
+```console
+woltspace dig grant next-colony my-ssh-alias --wolt n00b
+woltspace dig list
+woltspace dig connect next-colony
+woltspace dig connect next-colony -- woltspace status --json
+woltspace dig revoke next-colony
+```
+
+`grant` resolves `my-ssh-alias` through the owner's existing OpenSSH config and
+records the resulting host, user, and port. Every connection resolves it again
+and refuses a changed tuple. SSH runs with `StrictHostKeyChecking=yes`, so an
+unknown or changed server key fails instead of prompting the wolt to trust it.
+The destination alias is passed as an argument, never through a local shell.
+
+The default remote handoff location is `.woltspace/bootstrap`, relative to the
+remote SSH user's home. The command prints this location for the visiting wolt;
+v0 intentionally leaves the contents human-readable rather than imposing a
+protocol. A useful handoff includes what was installed, paths changed, checks
+run, open questions, and next steps. Existing files should be preserved unless
+the setup task explicitly authorizes replacing them.
+
+`revoke` removes Woltspace's local consent record. It cannot revoke the Unix
+account, SSH key, agent, or server-side authorization; the owner must remove
+those separately when access itself should end.
+
 ## Later: paired-colony visits
 
 Wire carries the request, destination approval, callback, and result. The
diff --git a/pyproject.toml b/pyproject.toml
index c732ca90..87c7703c 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -58,6 +58,7 @@ packages = ["src/woltspace"]
 "docs/updates.md" = "woltspace/_bundle/docs/updates.md"
 "docs/shared-skills.md" = "woltspace/_bundle/docs/shared-skills.md"
 "docs/colony-seeds.md" = "woltspace/_bundle/docs/colony-seeds.md"
+"docs/digging-v0.md" = "woltspace/_bundle/docs/digging-v0.md"
 
 [tool.hatch.build.targets.sdist]
 include = [
@@ -71,5 +72,6 @@ include = [
     "/docs/updates.md",
     "/docs/shared-skills.md",
     "/docs/colony-seeds.md",
+    "/docs/digging-v0.md",
     "/LICENSE",
 ]
diff --git a/src/woltspace/cli.py b/src/woltspace/cli.py
index 59c23221..bcd6db9e 100644
--- a/src/woltspace/cli.py
+++ b/src/woltspace/cli.py
@@ -8,7 +8,7 @@
 import sys
 
 from . import __version__, lore
-from .envvars import warn_legacy_once
+from .envvars import get_env, warn_legacy_once
 from .layout import RuntimeLayout
 
 
@@ -585,6 +585,103 @@ def _seed(args) -> int:
     return 1
 
 
+def _dig(args) -> int:
+    args.dig_parser.print_help()
+    return 1
+
+
+def _dig_store():
+    from .dig import DigStore
+
+    return DigStore(RuntimeLayout.from_env().state_root)
+
+
+def _dig_grant(args) -> int:
+    from .dig import DigError, resolve_ssh
+
+    layout = RuntimeLayout.from_env()
+    if not (layout.wolts_dir / args.wolt).is_dir():
+        lore.failure(f"dig grant failed: unknown wolt: {args.wolt}")
+        return 1
+    try:
+        resolved = resolve_ssh(args.destination)
+        grant = _dig_store().grant(
+            name=args.name,
+            wolt=args.wolt,
+            destination=args.destination,
+            resolved=resolved,
+            bootstrap_dir=args.bootstrap_dir,
+        )
+    except DigError as exc:
+        lore.failure(f"dig grant failed: {exc}")
+        return 1
+    if args.json:
+        print(json.dumps({"ok": True, "grant": grant}, indent=2))
+    else:
+        lore.headline(lore.TRACKS, f"dig approved: {grant['name']}")
+        lore.labelled("wolt", grant["wolt"])
+        lore.labelled("ssh", grant["destination"])
+        lore.labelled("resolved", f"{resolved.user}@{resolved.hostname}:{resolved.port}")
+        lore.labelled("handoff", grant["bootstrap_dir"])
+        lore.subtitle("this records consent; the SSH user's real permissions still apply")
+    return 0
+
+
+def _dig_list(args) -> int:
+    grants = sorted(_dig_store().list(), key=lambda item: item["name"])
+    if args.json:
+        print(json.dumps({"grants": grants}, indent=2))
+    elif not grants:
+        lore.headline(lore.TRACKS, "no approved digs")
+    else:
+        lore.headline(lore.TRACKS, f"approved digs: {len(grants)}")
+        for grant in grants:
+            lore.labelled(grant["name"], f"{grant['wolt']} -> {grant['destination']}")
+    return 0
+
+
+def _dig_revoke(args) -> int:
+    from .dig import DigError
+
+    try:
+        revoked = _dig_store().revoke(args.name)
+    except DigError as exc:
+        lore.failure(f"dig revoke failed: {exc}")
+        return 1
+    if args.json:
+        print(json.dumps({"ok": True, "revoked": revoked, "name": args.name}, indent=2))
+    else:
+        lore.headline(lore.MOON, f"dig {'revoked' if revoked else 'was not approved'}: {args.name}")
+        lore.subtitle("remove the SSH key/config separately if access itself must end")
+    return 0
+
+
+def _dig_connect(args) -> int:
+    from .dig import DigError, connect
+
+    try:
+        grant = _dig_store().get(args.name)
+        if not args.json:
+            lore.headline(lore.TRACKS, f"digging: {args.name}")
+            lore.labelled("wolt", grant["wolt"])
+            lore.labelled("handoff", grant["bootstrap_dir"])
+        remote_command = list(args.remote_command)
+        if remote_command[:1] == ["--"]:
+            remote_command = remote_command[1:]
+        return connect(
+            _dig_store(),
+            args.name,
+            command=remote_command,
+            actor_wolt=get_env("WOLTSPACE_WOLT_NAME", ""),
+        )
+    except DigError as exc:
+        if args.json:
+            print(json.dumps({"ok": False, "error": str(exc)}, indent=2))
+        else:
+            lore.failure(f"dig failed: {exc}")
+        return 1
+
+
 def _seed_create(args) -> int:
     from .seed import SeedError, create_seed
 
@@ -813,6 +910,33 @@ def build_parser() -> argparse.ArgumentParser:
     seed_install.add_argument("--json", action="store_true")
     seed_install.set_defaults(func=_seed_install)
 
+    dig = sub.add_parser("dig", help="let a wolt visit an owner-approved SSH host")
+    dig.set_defaults(func=_dig, dig_parser=dig)
+    dig_sub = dig.add_subparsers(dest="dig_command")
+
+    dig_grant = dig_sub.add_parser("grant", help="approve an exact SSH destination")
+    dig_grant.add_argument("name")
+    dig_grant.add_argument("destination", help="SSH host or alias from ~/.ssh/config")
+    dig_grant.add_argument("--wolt", required=True)
+    dig_grant.add_argument("--bootstrap-dir", default=".woltspace/bootstrap")
+    dig_grant.add_argument("--json", action="store_true")
+    dig_grant.set_defaults(func=_dig_grant)
+
+    dig_list = dig_sub.add_parser("list", help="show approved SSH destinations")
+    dig_list.add_argument("--json", action="store_true")
+    dig_list.set_defaults(func=_dig_list)
+
+    dig_revoke = dig_sub.add_parser("revoke", help="remove a Woltspace dig approval")
+    dig_revoke.add_argument("name")
+    dig_revoke.add_argument("--json", action="store_true")
+    dig_revoke.set_defaults(func=_dig_revoke)
+
+    dig_connect = dig_sub.add_parser("connect", help="open the approved SSH destination")
+    dig_connect.add_argument("name")
+    dig_connect.add_argument("remote_command", nargs=argparse.REMAINDER)
+    dig_connect.add_argument("--json", action="store_true")
+    dig_connect.set_defaults(func=_dig_connect)
+
     tui = sub.add_parser("tui", help="open the terminal UI")
     tui.add_argument("--dry-run", action="store_true", help="show resolution without launching")
     tui.add_argument("--json", action="store_true", help=argparse.SUPPRESS)
diff --git a/src/woltspace/dig.py b/src/woltspace/dig.py
new file mode 100644
index 00000000..db506482
--- /dev/null
+++ b/src/woltspace/dig.py
@@ -0,0 +1,239 @@
+"""Owner-approved SSH destinations for Woltspace digging v0."""
+
+from __future__ import annotations
+
+import fcntl
+import json
+import os
+import re
+import subprocess
+import time
+from contextlib import contextmanager
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Callable, Sequence
+
+
+STORE_VERSION = "woltspace.digs/v0"
+DEFAULT_BOOTSTRAP_DIR = ".woltspace/bootstrap"
+_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,62}$")
+_DESTINATION_RE = re.compile(r"^[A-Za-z0-9_.@:\[\]-]+$")
+_REMOTE_PATH_RE = re.compile(r"^[A-Za-z0-9._/-]+$")
+
+
+class DigError(ValueError):
+    pass
+
+
+@dataclass(frozen=True)
+class ResolvedSSH:
+    hostname: str
+    user: str
+    port: int
+
+    def to_record(self) -> dict:
+        return {"hostname": self.hostname, "user": self.user, "port": self.port}
+
+
+def _validate_name(name: str) -> str:
+    if not _NAME_RE.fullmatch(name):
+        raise DigError("dig name must use lowercase letters, numbers, '_' or '-'")
+    return name
+
+
+def _validate_destination(destination: str) -> str:
+    if not destination or destination.startswith("-") or not _DESTINATION_RE.fullmatch(destination):
+        raise DigError("SSH destination must be one host/alias without whitespace or options")
+    return destination
+
+
+def _validate_bootstrap_dir(value: str) -> str:
+    path = Path(value)
+    if (
+        not value
+        or not _REMOTE_PATH_RE.fullmatch(value)
+        or path.is_absolute()
+        or ".." in path.parts
+    ):
+        raise DigError("bootstrap directory must be relative to the remote user's home")
+    return value
+
+
+def resolve_ssh(
+    destination: str,
+    *,
+    runner: Callable[..., subprocess.CompletedProcess] = subprocess.run,
+) -> ResolvedSSH:
+    destination = _validate_destination(destination)
+    result = runner(
+        ["ssh", "-G", "--", destination],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if result.returncode:
+        detail = (result.stderr or "ssh configuration could not be resolved").strip()
+        raise DigError(detail)
+    values: dict[str, str] = {}
+    for line in result.stdout.splitlines():
+        key, separator, value = line.partition(" ")
+        if separator and key in {"hostname", "user", "port"} and key not in values:
+            values[key] = value.strip()
+    if not values.get("hostname") or not values.get("user"):
+        raise DigError("ssh -G did not return a hostname and user")
+    try:
+        port = int(values.get("port", "22"))
+    except ValueError as exc:
+        raise DigError("ssh -G returned an invalid port") from exc
+    if not 1 <= port <= 65535:
+        raise DigError("ssh -G returned an invalid port")
+    return ResolvedSSH(values["hostname"], values["user"], port)
+
+
+class DigStore:
+    def __init__(self, state_root: str | Path):
+        self.root = Path(state_root) / "digs"
+        self.path = self.root / "grants.json"
+        self.lock_path = self.root / "grants.lock"
+
+    @contextmanager
+    def _locked(self):
+        self.root.mkdir(parents=True, exist_ok=True, mode=0o700)
+        os.chmod(self.root, 0o700)
+        fd = os.open(self.lock_path, os.O_RDWR | os.O_CREAT, 0o600)
+        try:
+            fcntl.flock(fd, fcntl.LOCK_EX)
+            yield
+        finally:
+            os.close(fd)
+
+    def _read(self) -> dict:
+        if not self.path.exists():
+            return {"version": STORE_VERSION, "grants": []}
+        try:
+            payload = json.loads(self.path.read_text())
+        except (OSError, json.JSONDecodeError) as exc:
+            raise DigError("dig grant store is unreadable") from exc
+        if payload.get("version") != STORE_VERSION or not isinstance(payload.get("grants"), list):
+            raise DigError("dig grant store has an unsupported shape")
+        return payload
+
+    def _write(self, payload: dict) -> None:
+        temp = self.path.with_suffix(".tmp")
+        data = (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode()
+        fd = os.open(temp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+        try:
+            with os.fdopen(fd, "wb") as handle:
+                handle.write(data)
+                handle.flush()
+                os.fsync(handle.fileno())
+            os.replace(temp, self.path)
+        finally:
+            try:
+                temp.unlink()
+            except FileNotFoundError:
+                pass
+
+    def grant(
+        self,
+        *,
+        name: str,
+        wolt: str,
+        destination: str,
+        resolved: ResolvedSSH,
+        bootstrap_dir: str = DEFAULT_BOOTSTRAP_DIR,
+        now: int | None = None,
+    ) -> dict:
+        name = _validate_name(name)
+        destination = _validate_destination(destination)
+        bootstrap_dir = _validate_bootstrap_dir(bootstrap_dir)
+        now = int(time.time()) if now is None else int(now)
+        record = {
+            "name": name,
+            "wolt": wolt,
+            "destination": destination,
+            "resolved": resolved.to_record(),
+            "bootstrap_dir": bootstrap_dir,
+            "created_at": now,
+            "connect_count": 0,
+            "last_connected_at": None,
+            "last_exit_code": None,
+        }
+        with self._locked():
+            payload = self._read()
+            if any(item.get("name") == name for item in payload["grants"]):
+                raise DigError(f"dig already exists: {name}")
+            payload["grants"].append(record)
+            self._write(payload)
+        return dict(record)
+
+    def get(self, name: str) -> dict:
+        _validate_name(name)
+        with self._locked():
+            for record in self._read()["grants"]:
+                if record.get("name") == name:
+                    return dict(record)
+        raise DigError(f"unknown dig: {name}")
+
+    def list(self) -> list[dict]:
+        with self._locked():
+            return [dict(item) for item in self._read()["grants"]]
+
+    def revoke(self, name: str) -> bool:
+        _validate_name(name)
+        with self._locked():
+            payload = self._read()
+            kept = [item for item in payload["grants"] if item.get("name") != name]
+            if len(kept) == len(payload["grants"]):
+                return False
+            payload["grants"] = kept
+            self._write(payload)
+            return True
+
+    def record_connection(self, name: str, *, exit_code: int, now: int | None = None) -> None:
+        now = int(time.time()) if now is None else int(now)
+        with self._locked():
+            payload = self._read()
+            for record in payload["grants"]:
+                if record.get("name") == name:
+                    record["connect_count"] = int(record.get("connect_count", 0)) + 1
+                    record["last_connected_at"] = now
+                    record["last_exit_code"] = int(exit_code)
+                    self._write(payload)
+                    return
+        raise DigError(f"unknown dig: {name}")
+
+
+def connect(
+    store: DigStore,
+    name: str,
+    *,
+    command: Sequence[str] = (),
+    actor_wolt: str = "",
+    resolver: Callable[[str], ResolvedSSH] = resolve_ssh,
+    runner: Callable[..., subprocess.CompletedProcess] = subprocess.run,
+) -> int:
+    grant = store.get(name)
+    if actor_wolt and actor_wolt != grant["wolt"]:
+        raise DigError(f"dig belongs to wolt {grant['wolt']}, not {actor_wolt}")
+    current = resolver(grant["destination"])
+    if current.to_record() != grant["resolved"]:
+        raise DigError("SSH destination no longer resolves to the approved host, user, and port")
+    argv = [
+        "ssh",
+        "-o", "StrictHostKeyChecking=yes",
+        "--",
+        grant["destination"],
+    ]
+    if command:
+        argv.extend(command)
+    try:
+        result = runner(argv, check=False)
+    except KeyboardInterrupt:
+        store.record_connection(name, exit_code=130)
+        raise
+    except OSError as exc:
+        store.record_connection(name, exit_code=126)
+        raise DigError(f"could not start ssh: {exc}") from exc
+    store.record_connection(name, exit_code=result.returncode)
+    return int(result.returncode)
diff --git a/test/test_dig_cli.py b/test/test_dig_cli.py
new file mode 100644
index 00000000..57bb0b4a
--- /dev/null
+++ b/test/test_dig_cli.py
@@ -0,0 +1,141 @@
+import json
+import subprocess
+
+import pytest
+
+from woltspace.dig import DigError, DigStore, ResolvedSSH, connect, resolve_ssh
+
+
+def resolved(host="server.example", user="colony", port=22):
+    return ResolvedSSH(host, user, port)
+
+
+def grant(store):
+    return store.grant(
+        name="new-colony",
+        wolt="n00b",
+        destination="my-colony",
+        resolved=resolved(),
+        now=100,
+    )
+
+
+def test_resolve_ssh_uses_config_without_a_shell():
+    calls = []
+
+    def runner(argv, **kwargs):
+        calls.append((argv, kwargs))
+        return subprocess.CompletedProcess(
+            argv, 0, "hostname server.example\nuser colony\nport 2222\n", ""
+        )
+
+    assert resolve_ssh("my-colony", runner=runner) == resolved(port=2222)
+    assert calls[0][0] == ["ssh", "-G", "--", "my-colony"]
+
+
+@pytest.mark.parametrize("destination", ["-oProxyCommand=oops", "host name", "host;id", ""])
+def test_destination_cannot_inject_ssh_options(destination):
+    with pytest.raises(DigError):
+        resolve_ssh(destination, runner=lambda *a, **k: None)
+
+
+def test_grant_store_is_private_and_revoke_is_honest(tmp_path):
+    store = DigStore(tmp_path / ".space")
+    record = grant(store)
+    assert record["bootstrap_dir"] == ".woltspace/bootstrap"
+    assert store.root.stat().st_mode & 0o777 == 0o700
+    assert store.path.stat().st_mode & 0o777 == 0o600
+    assert store.revoke("new-colony") is True
+    assert store.revoke("new-colony") is False
+    with pytest.raises(DigError, match="unknown dig"):
+        store.get("new-colony")
+
+
+def test_connect_rechecks_resolution_uses_strict_host_key_and_audits(tmp_path):
+    store = DigStore(tmp_path / ".space")
+    grant(store)
+    calls = []
+
+    def runner(argv, **kwargs):
+        calls.append((argv, kwargs))
+        return subprocess.CompletedProcess(argv, 7)
+
+    code = connect(
+        store,
+        "new-colony",
+        command=("woltspace", "status", "--json"),
+        resolver=lambda _: resolved(),
+        runner=runner,
+    )
+    assert code == 7
+    assert calls == [([
+        "ssh", "-o", "StrictHostKeyChecking=yes", "--", "my-colony",
+        "woltspace", "status", "--json",
+    ], {"check": False})]
+    audited = store.get("new-colony")
+    assert audited["connect_count"] == 1
+    assert audited["last_exit_code"] == 7
+
+
+def test_connect_refuses_changed_ssh_resolution_before_running(tmp_path):
+    store = DigStore(tmp_path / ".space")
+    grant(store)
+    ran = False
+
+    def runner(*args, **kwargs):
+        nonlocal ran
+        ran = True
+
+    with pytest.raises(DigError, match="no longer resolves"):
+        connect(store, "new-colony", resolver=lambda _: resolved(host="evil.example"), runner=runner)
+    assert ran is False
+    assert store.get("new-colony")["connect_count"] == 0
+
+
+def test_session_wolt_cannot_use_another_wolts_dig(tmp_path):
+    store = DigStore(tmp_path / ".space")
+    grant(store)
+    with pytest.raises(DigError, match="belongs to wolt n00b"):
+        connect(
+            store,
+            "new-colony",
+            actor_wolt="someone-else",
+            resolver=lambda _: resolved(),
+            runner=lambda *a, **k: None,
+        )
+
+
+def test_interrupted_connection_is_audited(tmp_path):
+    store = DigStore(tmp_path / ".space")
+    grant(store)
+
+    def interrupted(*args, **kwargs):
+        raise KeyboardInterrupt
+
+    with pytest.raises(KeyboardInterrupt):
+        connect(store, "new-colony", resolver=lambda _: resolved(), runner=interrupted)
+    audit = store.get("new-colony")
+    assert audit["connect_count"] == 1
+    assert audit["last_exit_code"] == 130
+
+
+def test_bootstrap_path_cannot_escape_remote_home(tmp_path):
+    store = DigStore(tmp_path / ".space")
+    with pytest.raises(DigError, match="relative"):
+        store.grant(
+            name="bad",
+            wolt="n00b",
+            destination="host",
+            resolved=resolved(),
+            bootstrap_dir="../../etc",
+        )
+
+
+def test_store_contains_no_credentials_or_command_bodies(tmp_path):
+    store = DigStore(tmp_path / ".space")
+    grant(store)
+    payload = json.loads(store.path.read_text())
+    text = json.dumps(payload)
+    assert "PRIVATE KEY" not in text
+    assert "remote_command" not in text
+    assert payload["version"] == "woltspace.digs/v0"

From 9691e5ae274c74c9f8e6436dc1f409e29df3380d Mon Sep 17 00:00:00 2001
From: "woltspace-jerpint[bot]"
 <268897999+woltspace-jerpint[bot]@users.noreply.github.com>
Date: Sun, 20 Sep 2026 08:42:23 -0400
Subject: [PATCH 06/10] feat: teach wolts the digging workflow

---
 container/skills/digging/SKILL.md | 79 +++++++++++++++++++++++++++++++
 1 file changed, 79 insertions(+)
 create mode 100644 container/skills/digging/SKILL.md

diff --git a/container/skills/digging/SKILL.md b/container/skills/digging/SKILL.md
new file mode 100644
index 00000000..630317c2
--- /dev/null
+++ b/container/skills/digging/SKILL.md
@@ -0,0 +1,79 @@
+---
+name: digging
+description: Use owner-approved SSH access to inspect or set up a remote machine, bootstrap a Woltspace colony, and leave a durable local handoff. Use when a human asks a wolt to SSH into, dig into, configure, or prepare another machine. Do not use for Wire pairing or cross-colony IWCL.
+---
+
+# Dig into an owned machine
+
+Digging means temporarily using the owner's existing SSH access. The wolt stays
+resident in its home colony: connect, do the authorized work, leave a local
+handoff for future wolts on that machine, disconnect, and report home.
+
+## Find or create the approval
+
+Run `woltspace dig list --json`. Use an existing grant only when its wolt and
+destination match the human's request.
+
+If none matches, an explicit instruction such as “dig into `newbox` and set up
+the colony” authorizes creating that local grant for the current wolt when
+`newbox` is an unambiguous existing SSH alias:
+
+```sh
+woltspace dig grant newbox newbox --wolt "$WOLTSPACE_WOLT_NAME"
+```
+
+If the destination, wolt, or intended work is ambiguous, ask before granting or
+connecting. Never guess a hostname or broaden an instruction about one machine
+to another.
+
+The grant records Woltspace consent; the SSH user's actual permissions remain
+the authority. Do not create accounts, keys, tunnels, or broader server access
+unless the human separately asks for that work.
+
+## Connect safely
+
+Start with a small read-only probe appropriate to the request, for example:
+
+```sh
+woltspace dig connect newbox -- hostname
+woltspace dig connect newbox -- whoami
+woltspace dig connect newbox -- command -v woltspace
+```
+
+Use `woltspace dig connect NAME -- COMMAND...` for bounded commands and
+`woltspace dig connect NAME` only when an interactive shell is genuinely useful.
+The command rechecks the approved SSH host, user, and port and requires strict
+host-key verification.
+
+Never disable `StrictHostKeyChecking`, accept a new host key on the human's
+behalf, copy a private SSH key, or send credentials through chat, Wire, command
+arguments, or handoff files. If SSH authentication or host verification fails,
+report the exact non-secret failure and let the human repair their normal SSH
+configuration.
+
+## Bootstrap and hand off
+
+Make changes only within the task the human authorized. Preserve existing data
+and inspect before overwriting configuration.
+
+The grant's `bootstrap_dir` is relative to the remote SSH user's home and
+defaults to `.woltspace/bootstrap`. Create it when needed and leave concise,
+human-readable artifacts for the future colony, such as:
+
+- what was installed or configured;
+- important paths and decisions;
+- checks run and their results;
+- incomplete work, open questions, and safe next steps.
+
+Do not write into another wolt's private memory. Do not copy the visiting
+wolt's identity, memories, sessions, credentials, or local configuration to the
+remote machine. A colony seed may be installed there only when the human has
+selected and authorized that seed; digging permission alone does not select one.
+
+## Return and report
+
+Disconnect when the requested work is complete or progress is blocked. Report
+the destination, changes, verification, handoff path, and anything left undone.
+Do not claim that `woltspace dig revoke NAME` removes real SSH access: it removes
+only Woltspace's local grant. The owner must separately remove SSH keys,
+accounts, or server authorization when underlying access should end.

From c470bc6321ac018aa252da59b83d3cb455e57aca Mon Sep 17 00:00:00 2001
From: "woltspace-jerpint[bot]"
 <268897999+woltspace-jerpint[bot]@users.noreply.github.com>
Date: Sun, 20 Sep 2026 08:45:36 -0400
Subject: [PATCH 07/10] fix: make digging consent explicit and discoverable

---
 container/skills/digging/SKILL.md | 30 ++++++++++++++++++++++++------
 1 file changed, 24 insertions(+), 6 deletions(-)

diff --git a/container/skills/digging/SKILL.md b/container/skills/digging/SKILL.md
index 630317c2..d88c59c9 100644
--- a/container/skills/digging/SKILL.md
+++ b/container/skills/digging/SKILL.md
@@ -9,22 +9,29 @@ Digging means temporarily using the owner's existing SSH access. The wolt stays
 resident in its home colony: connect, do the authorized work, leave a local
 handoff for future wolts on that machine, disconnect, and report home.
 
+Prefer `woltspace dig` over invoking `ssh` directly whenever the digging command
+is available. The wrapper makes the intended wolt and destination visible,
+rechecks the resolved target, preserves the handoff convention, and records a
+small non-secret audit trail. It does not create a security boundary against a
+wolt running as the same Unix user.
+
 ## Find or create the approval
 
 Run `woltspace dig list --json`. Use an existing grant only when its wolt and
 destination match the human's request.
 
-If none matches, an explicit instruction such as “dig into `newbox` and set up
-the colony” authorizes creating that local grant for the current wolt when
-`newbox` is an unambiguous existing SSH alias:
+If none matches and the named destination is an unambiguous existing SSH alias,
+set up the dig grant yourself rather than asking the human to remember CLI
+syntax. Choose a short lowercase grant name and run:
 
 ```sh
 woltspace dig grant newbox newbox --wolt "$WOLTSPACE_WOLT_NAME"
 ```
 
-If the destination, wolt, or intended work is ambiguous, ask before granting or
-connecting. Never guess a hostname or broaden an instruction about one machine
-to another.
+Creating the grant only records a pointer to existing SSH access; it copies no
+key, certificate, agent credential, or token. If the destination, wolt, or
+intended work is ambiguous, ask before granting. Never guess a hostname or
+browse unrelated SSH destinations looking for somewhere to connect.
 
 The grant records Woltspace consent; the SSH user's actual permissions remain
 the authority. Do not create accounts, keys, tunnels, or broader server access
@@ -32,6 +39,17 @@ unless the human separately asks for that work.
 
 ## Connect safely
 
+Immediately before the first connection in a task, show the human the grant
+name and resolved `user@host:port`, briefly state the intended work, and ask for
+confirmation. A stored grant is not standing permission to connect.
+
+Skip that confirmation only when the human's current instruction explicitly
+says that permission is not required, says to proceed without asking, or gives
+equally clear authorization to connect immediately. Do not infer a permanent
+waiver from an earlier task or from the mere existence of SSH access. Once the
+human confirms a task, do not repeatedly ask for every bounded command needed
+to complete that same task unless the target or scope changes.
+
 Start with a small read-only probe appropriate to the request, for example:
 
 ```sh

From c2602e9b9b34f67206a702ec51c94bc3e9a9d683 Mon Sep 17 00:00:00 2001
From: "woltspace-jerpint[bot]"
 <268897999+woltspace-jerpint[bot]@users.noreply.github.com>
Date: Sun, 20 Sep 2026 08:58:40 -0400
Subject: [PATCH 08/10] docs: outline cloudflare transport for future digs

---
 docs/digging-v0.md | 21 +++++++++++++++++++++
 1 file changed, 21 insertions(+)

diff --git a/docs/digging-v0.md b/docs/digging-v0.md
index e24c874c..ec612c0d 100644
--- a/docs/digging-v0.md
+++ b/docs/digging-v0.md
@@ -87,6 +87,27 @@ Cross-tunnel IWCL, a visiting-wolt identity bridge, callbacks, and ongoing
 collaboration are explicitly out of v0. Real digging experience should tell us
 whether they are needed and what identity and consent model they require.
 
+### Candidate v1 transport: Cloudflare
+
+For machine-to-machine digging, a destination-owned `cloudflared` connector can
+reach Cloudflare over outbound-only connections, avoiding a public origin IP or
+inbound router port. The durable tunnel is transport; per-dig authorization is
+the ephemeral part.
+
+Two increments are possible:
+
+1. Reuse owner-managed SSH keys through a private Cloudflare Tunnel/WARP route.
+2. Use Cloudflare Access for Infrastructure for short-lived SSH certificates,
+   exact user/port policy, and access or command auditing.
+
+The destination owns its tunnel token and never sends it to the visiting wolt.
+Direct SSH exposure must still be blocked at the origin if Cloudflare-only
+access is intended. The legacy Cloudflare short-lived-certificate application
+flow is not a new-deployment target; evaluate Access for Infrastructure instead.
+
+This is unnecessary for two local users on one Mac. A future non-network
+local-user transport would be a smaller solution for that case.
+
 ## V1 human experience
 
 1. Alice asks to dig to Bobeaver Colony with a short task description and a

From bb9d5707dfe823aa6b99775dc57bf1b9aab714e4 Mon Sep 17 00:00:00 2001
From: "woltspace-jerpint[bot]"
 <268897999+woltspace-jerpint[bot]@users.noreply.github.com>
Date: Sun, 20 Sep 2026 09:05:24 -0400
Subject: [PATCH 09/10] chore: keep digging change scoped to ssh v0

---
 container/lib/digging.py | 268 ---------------------------------------
 docs/digging-v0.md       |  18 +--
 test/test_digging.py     | 136 --------------------
 3 files changed, 6 insertions(+), 416 deletions(-)
 delete mode 100644 container/lib/digging.py
 delete mode 100644 test/test_digging.py

diff --git a/container/lib/digging.py b/container/lib/digging.py
deleted file mode 100644
index 1607370e..00000000
--- a/container/lib/digging.py
+++ /dev/null
@@ -1,268 +0,0 @@
-"""Destination-owned authorization state for experimental cross-colony digs.
-
-Wire authenticates and encrypts messages.  This module deliberately does not:
-it consumes the peer identity already verified by a transport adapter and owns
-the separate Woltspace decision about whether that peer may start one guest
-session.
-
-The first proof is intentionally only an authorization kernel.  It neither
-opens a network listener nor starts SSH.  A later session adapter may consume
-an approved grant exactly once and map its target into a disposable workspace.
-"""
-
-from __future__ import annotations
-
-import hashlib
-import hmac
-import json
-import os
-import secrets
-import time
-from contextlib import contextmanager
-from dataclasses import asdict, dataclass
-from pathlib import Path
-
-
-DIG_VERSION = "woltspace-dig/v0"
-MAX_LIFETIME_SECONDS = 15 * 60
-_ID_BYTES = 16
-_CAP_BYTES = 32
-_STATES = frozenset({"pending", "approved", "active", "completed", "revoked", "expired"})
-
-
-class DigError(ValueError):
-    """An invalid request or lifecycle transition."""
-
-
-@dataclass(frozen=True)
-class DigRequest:
-    version: str
-    request_id: str
-    source_colony: str
-    source_wolt: str
-    task: str
-    target: str
-    created_at: int
-    expires_at: int
-
-    @classmethod
-    def create(
-        cls,
-        *,
-        source_colony: str,
-        source_wolt: str,
-        task: str,
-        target: str,
-        now: int | None = None,
-        lifetime_seconds: int = 10 * 60,
-    ) -> "DigRequest":
-        now = int(time.time()) if now is None else int(now)
-        if not 1 <= lifetime_seconds <= MAX_LIFETIME_SECONDS:
-            raise DigError("dig request lifetime must be between 1 and 900 seconds")
-        values = {
-            "source_colony": source_colony,
-            "source_wolt": source_wolt,
-            "task": task,
-            "target": target,
-        }
-        for name, value in values.items():
-            if not isinstance(value, str) or not value.strip():
-                raise DigError(f"{name} must be a non-empty string")
-        if len(task.encode("utf-8")) > 500:
-            raise DigError("task exceeds 500 UTF-8 bytes")
-        if target.startswith("/") or ".." in Path(target).parts:
-            raise DigError("target must be a destination-relative path")
-        return cls(
-            version=DIG_VERSION,
-            request_id=secrets.token_hex(_ID_BYTES),
-            source_colony=source_colony,
-            source_wolt=source_wolt,
-            task=task,
-            target=target,
-            created_at=now,
-            expires_at=now + lifetime_seconds,
-        )
-
-    @classmethod
-    def from_dict(cls, value: dict) -> "DigRequest":
-        if not isinstance(value, dict) or set(value) != set(cls.__dataclass_fields__):
-            raise DigError("dig request has unknown or missing fields")
-        request = cls(**value)
-        if request.version != DIG_VERSION:
-            raise DigError("unsupported dig request version")
-        for name in ("request_id", "source_colony", "source_wolt", "task", "target"):
-            item = getattr(request, name)
-            if not isinstance(item, str) or not item.strip():
-                raise DigError(f"{name} must be a non-empty string")
-        if len(request.request_id) != _ID_BYTES * 2 or any(
-            c not in "0123456789abcdef" for c in request.request_id
-        ):
-            raise DigError("invalid dig request id")
-        if len(request.task.encode("utf-8")) > 500:
-            raise DigError("task exceeds 500 UTF-8 bytes")
-        if request.target.startswith("/") or ".." in Path(request.target).parts:
-            raise DigError("target must be a destination-relative path")
-        if type(request.created_at) is not int or type(request.expires_at) is not int:
-            raise DigError("dig request times must be integers")
-        if request.expires_at <= request.created_at:
-            raise DigError("dig request expiry must follow creation")
-        if request.expires_at - request.created_at > MAX_LIFETIME_SECONDS:
-            raise DigError("dig request lifetime exceeds 900 seconds")
-        return request
-
-    def to_dict(self) -> dict:
-        return asdict(self)
-
-
-class DigStore:
-    """Private destination-side store for pending requests and one-use grants."""
-
-    def __init__(self, root: str | Path):
-        self.root = Path(root)
-        self.root.mkdir(parents=True, exist_ok=True, mode=0o700)
-        os.chmod(self.root, 0o700)
-
-    def _path(self, request_id: str) -> Path:
-        if len(request_id) != _ID_BYTES * 2 or any(c not in "0123456789abcdef" for c in request_id):
-            raise DigError("invalid dig request id")
-        return self.root / f"{request_id}.json"
-
-    @contextmanager
-    def _locked(self, request_id: str):
-        lock_path = self.root / f".{request_id}.lock"
-        fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600)
-        try:
-            import fcntl
-
-            fcntl.flock(fd, fcntl.LOCK_EX)
-            yield self._path(request_id)
-        finally:
-            os.close(fd)
-
-    @staticmethod
-    def _cap_digest(capability: str) -> str:
-        return hashlib.sha256(capability.encode("ascii")).hexdigest()
-
-    def _write_new(self, path: Path, record: dict) -> None:
-        payload = (json.dumps(record, sort_keys=True, separators=(",", ":")) + "\n").encode()
-        fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
-        with os.fdopen(fd, "wb") as handle:
-            handle.write(payload)
-
-    def _replace(self, path: Path, record: dict) -> None:
-        temp = path.with_suffix(f".{secrets.token_hex(8)}.tmp")
-        payload = (json.dumps(record, sort_keys=True, separators=(",", ":")) + "\n").encode()
-        fd = os.open(temp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
-        try:
-            with os.fdopen(fd, "wb") as handle:
-                handle.write(payload)
-                handle.flush()
-                os.fsync(handle.fileno())
-            os.replace(temp, path)
-        finally:
-            try:
-                temp.unlink()
-            except FileNotFoundError:
-                pass
-
-    def receive(self, request: DigRequest, *, verified_peer: str, now: int | None = None) -> dict:
-        now = int(time.time()) if now is None else int(now)
-        if verified_peer != request.source_colony:
-            raise DigError("Wire peer does not match dig request source")
-        if now >= request.expires_at:
-            raise DigError("dig request is expired")
-        record = {"request": request.to_dict(), "state": "pending", "received_at": now}
-        self._write_new(self._path(request.request_id), record)
-        return self.status(request.request_id, now=now)
-
-    def approve(self, request_id: str, *, now: int | None = None, lifetime_seconds: int = 5 * 60) -> str:
-        now = int(time.time()) if now is None else int(now)
-        if not 1 <= lifetime_seconds <= MAX_LIFETIME_SECONDS:
-            raise DigError("dig grant lifetime must be between 1 and 900 seconds")
-        with self._locked(request_id) as path:
-            record = self._load(path)
-            self._expire(record, now)
-            if record["state"] != "pending":
-                raise DigError(f"cannot approve dig in state {record['state']}")
-            capability = secrets.token_urlsafe(_CAP_BYTES)
-            record.update(
-                state="approved",
-                approved_at=now,
-                grant_expires_at=min(now + lifetime_seconds, record["request"]["expires_at"]),
-                capability_sha256=self._cap_digest(capability),
-            )
-            self._replace(path, record)
-            return capability
-
-    def consume(self, request_id: str, capability: str, *, verified_peer: str, now: int | None = None) -> dict:
-        now = int(time.time()) if now is None else int(now)
-        with self._locked(request_id) as path:
-            record = self._load(path)
-            self._expire(record, now)
-            if record["state"] != "approved":
-                if record["state"] == "active":
-                    raise DigError("dig grant was already consumed")
-                raise DigError(f"cannot consume dig in state {record['state']}")
-            if verified_peer != record["request"]["source_colony"]:
-                raise DigError("Wire peer does not match approved source")
-            supplied = self._cap_digest(capability)
-            if not hmac.compare_digest(supplied, record["capability_sha256"]):
-                raise DigError("invalid dig capability")
-            record.update(state="active", started_at=now)
-            record.pop("capability_sha256", None)
-            self._replace(path, record)
-            return self._public(record)
-
-    def finish(self, request_id: str, *, now: int | None = None) -> dict:
-        now = int(time.time()) if now is None else int(now)
-        with self._locked(request_id) as path:
-            record = self._load(path)
-            if record["state"] != "active":
-                raise DigError(f"cannot finish dig in state {record['state']}")
-            record.update(state="completed", finished_at=now)
-            self._replace(path, record)
-            return self._public(record)
-
-    def revoke(self, request_id: str, *, now: int | None = None) -> dict:
-        now = int(time.time()) if now is None else int(now)
-        with self._locked(request_id) as path:
-            record = self._load(path)
-            if record["state"] in {"completed", "revoked", "expired"}:
-                raise DigError(f"cannot revoke dig in state {record['state']}")
-            record.update(state="revoked", revoked_at=now)
-            record.pop("capability_sha256", None)
-            self._replace(path, record)
-            return self._public(record)
-
-    def status(self, request_id: str, *, now: int | None = None) -> dict:
-        now = int(time.time()) if now is None else int(now)
-        with self._locked(request_id) as path:
-            record = self._load(path)
-            if self._expire(record, now):
-                self._replace(path, record)
-            return self._public(record)
-
-    @staticmethod
-    def _public(record: dict) -> dict:
-        public = dict(record)
-        public.pop("capability_sha256", None)
-        return public
-
-    @staticmethod
-    def _load(path: Path) -> dict:
-        try:
-            record = json.loads(path.read_text())
-        except FileNotFoundError as exc:
-            raise DigError("unknown dig request") from exc
-        if record.get("state") not in _STATES:
-            raise DigError("invalid stored dig state")
-        return record
-
-    @staticmethod
-    def _expire(record: dict, now: int) -> bool:
-        deadline = record.get("grant_expires_at", record["request"]["expires_at"])
-        if record["state"] in {"pending", "approved"} and now >= deadline:
-            record.update(state="expired", expired_at=now)
-            record.pop("capability_sha256", None)
-            return True
-        return False
diff --git a/docs/digging-v0.md b/docs/digging-v0.md
index ec612c0d..43b94380 100644
--- a/docs/digging-v0.md
+++ b/docs/digging-v0.md
@@ -80,8 +80,8 @@ those separately when access itself should end.
 
 ## Later: paired-colony visits
 
-Wire carries the request, destination approval, callback, and result. The
-destination-owned authorization kernel in this branch belongs here.
+Wire could later carry the request, destination approval, callback, and result.
+That authorization layer is deliberately not part of this SSH v0.
 
 Cross-tunnel IWCL, a visiting-wolt identity bridge, callbacks, and ongoing
 collaboration are explicitly out of v0. Real digging experience should tell us
@@ -108,7 +108,7 @@ flow is not a new-deployment target; evaluate Access for Infrastructure instead.
 This is unnecessary for two local users on one Mac. A future non-network
 local-user transport would be a smaller solution for that case.
 
-## V1 human experience
+## Possible V1 human experience
 
 1. Alice asks to dig to Bobeaver Colony with a short task description and a
    destination-relative disposable workspace.
@@ -149,12 +149,6 @@ a prerequisite for the general digging concept.
 
 ## Current implementation
 
-`container/lib/digging.py` provides strict request parsing and a private durable
-state machine: `pending -> approved -> active -> completed`, with `revoked` and
-`expired` terminal paths. Capabilities are stored only as SHA-256 digests and
-removed on redemption. `verified_peer` is an input from the future Wire adapter,
-not a claim accepted from the request body.
-
-This is not yet a usable remote-access feature. The next implementation gate is
-the simpler SSH v0. The Wire state machine remains isolated V1 groundwork until
-the SSH setup/access experience and its audit boundary are proven.
+SSH v0 is the only implemented path. Wire authorization, cross-tunnel IWCL,
+local-user switching, Cloudflare transport, and short-lived infrastructure
+credentials remain design notes, not shipped capability.
diff --git a/test/test_digging.py b/test/test_digging.py
deleted file mode 100644
index be61f444..00000000
--- a/test/test_digging.py
+++ /dev/null
@@ -1,136 +0,0 @@
-import json
-import sys
-import threading
-from pathlib import Path
-
-import pytest
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "container" / "lib"))
-
-from digging import DIG_VERSION, DigError, DigRequest, DigStore
-
-
-def request(now=100):
-    return DigRequest.create(
-        source_colony="alice-root-key",
-        source_wolt="n00b",
-        task="Create a hello file in the disposable guest workspace",
-        target="guest-work",
-        now=now,
-        lifetime_seconds=600,
-    )
-
-
-def test_owned_colony_happy_path_is_explicit_and_one_use(tmp_path):
-    store = DigStore(tmp_path / "digs")
-    req = request()
-    pending = store.receive(req, verified_peer="alice-root-key", now=101)
-    assert pending["state"] == "pending"
-
-    capability = store.approve(req.request_id, now=102, lifetime_seconds=120)
-    assert capability not in json.dumps(store.status(req.request_id, now=103))
-    active = store.consume(
-        req.request_id, capability, verified_peer="alice-root-key", now=104
-    )
-    assert active["state"] == "active"
-    with pytest.raises(DigError, match="already consumed"):
-        store.consume(req.request_id, capability, verified_peer="alice-root-key", now=105)
-
-    assert store.finish(req.request_id, now=106)["state"] == "completed"
-
-
-def test_wire_pairing_does_not_implicitly_authorize_a_dig(tmp_path):
-    store = DigStore(tmp_path / "digs")
-    req = request()
-    store.receive(req, verified_peer="alice-root-key", now=101)
-    with pytest.raises(DigError, match="state pending"):
-        store.consume(req.request_id, "anything", verified_peer="alice-root-key", now=102)
-
-
-def test_transport_peer_must_match_claimed_source(tmp_path):
-    store = DigStore(tmp_path / "digs")
-    with pytest.raises(DigError, match="does not match"):
-        store.receive(request(), verified_peer="mallory-root-key", now=101)
-
-
-def test_grant_is_bound_to_source_peer(tmp_path):
-    store = DigStore(tmp_path / "digs")
-    req = request()
-    store.receive(req, verified_peer="alice-root-key", now=101)
-    capability = store.approve(req.request_id, now=102)
-    with pytest.raises(DigError, match="does not match"):
-        store.consume(req.request_id, capability, verified_peer="mallory-root-key", now=103)
-
-
-def test_expired_or_revoked_grants_cannot_be_consumed(tmp_path):
-    store = DigStore(tmp_path / "digs")
-    expired = request()
-    store.receive(expired, verified_peer="alice-root-key", now=101)
-    cap = store.approve(expired.request_id, now=102, lifetime_seconds=2)
-    with pytest.raises(DigError, match="state expired"):
-        store.consume(expired.request_id, cap, verified_peer="alice-root-key", now=104)
-
-    revoked = request(now=200)
-    store.receive(revoked, verified_peer="alice-root-key", now=201)
-    cap = store.approve(revoked.request_id, now=202)
-    assert store.revoke(revoked.request_id, now=203)["state"] == "revoked"
-    with pytest.raises(DigError, match="state revoked"):
-        store.consume(revoked.request_id, cap, verified_peer="alice-root-key", now=204)
-
-
-def test_request_schema_and_target_are_strict():
-    req = request()
-    payload = req.to_dict()
-    payload["surprise"] = True
-    with pytest.raises(DigError, match="unknown or missing"):
-        DigRequest.from_dict(payload)
-    with pytest.raises(DigError, match="destination-relative"):
-        DigRequest.create(
-            source_colony="alice",
-            source_wolt="n00b",
-            task="nope",
-            target="../host",
-            now=1,
-        )
-    assert req.version == DIG_VERSION
-
-
-def test_store_is_private_and_never_persists_raw_capability(tmp_path):
-    store = DigStore(tmp_path / "digs")
-    req = request()
-    store.receive(req, verified_peer="alice-root-key", now=101)
-    capability = store.approve(req.request_id, now=102)
-    path = tmp_path / "digs" / f"{req.request_id}.json"
-    assert (tmp_path / "digs").stat().st_mode & 0o777 == 0o700
-    assert path.stat().st_mode & 0o777 == 0o600
-    assert capability not in path.read_text()
-
-
-def test_only_one_concurrent_redeemer_wins(tmp_path):
-    store = DigStore(tmp_path / "digs")
-    req = request()
-    store.receive(req, verified_peer="alice-root-key", now=101)
-    capability = store.approve(req.request_id, now=102)
-    barrier = threading.Barrier(12)
-    outcomes = []
-
-    def redeem():
-        barrier.wait()
-        try:
-            store.consume(
-                req.request_id,
-                capability,
-                verified_peer="alice-root-key",
-                now=103,
-            )
-            outcomes.append("won")
-        except DigError:
-            outcomes.append("lost")
-
-    workers = [threading.Thread(target=redeem) for _ in range(12)]
-    for worker in workers:
-        worker.start()
-    for worker in workers:
-        worker.join()
-    assert outcomes.count("won") == 1
-    assert outcomes.count("lost") == 11

From 6ac02b82008c436948122d37ad8e865f06c53a97 Mon Sep 17 00:00:00 2001
From: "woltspace-jerpint[bot]"
 <268897999+woltspace-jerpint[bot]@users.noreply.github.com>
Date: Sun, 20 Sep 2026 09:15:20 -0400
Subject: [PATCH 10/10] test: prove digging through disposable ssh colony

---
 docs/digging-v0.md                    |  15 ++
 src/woltspace/dig.py                  |  16 +-
 test/e2e/dig_ssh_container/Dockerfile |  26 ++++
 test/e2e/dig_ssh_container/run.py     | 213 ++++++++++++++++++++++++++
 test/test_dig_cli.py                  |  37 ++++-
 5 files changed, 304 insertions(+), 3 deletions(-)
 create mode 100644 test/e2e/dig_ssh_container/Dockerfile
 create mode 100644 test/e2e/dig_ssh_container/run.py

diff --git a/docs/digging-v0.md b/docs/digging-v0.md
index 43b94380..9de697c3 100644
--- a/docs/digging-v0.md
+++ b/docs/digging-v0.md
@@ -78,6 +78,21 @@ the setup task explicitly authorizes replacing them.
 account, SSH key, agent, or server-side authorization; the owner must remove
 those separately when access itself should end.
 
+### Disposable installed-wheel proof
+
+Reviewers with Docker can run the complete loopback-only SSH proof:
+
+```console
+uv run python test/e2e/dig_ssh_container/run.py
+```
+
+It builds the candidate wheel, installs it on both client and disposable SSH
+target, publishes the target only on a random `127.0.0.1` port, uses isolated
+temporary keys/config/known-hosts/state, pins the generated host key, and proves
+grant, installed CLI connection, a separate Unix home, colony bootstrap files,
+handoff persistence after reconnect, audit, revoke, refusal after revoke, and
+cleanup. It never uses the host's normal SSH configuration or Woltspace state.
+
 ## Later: paired-colony visits
 
 Wire could later carry the request, destination approval, callback, and result.
diff --git a/src/woltspace/dig.py b/src/woltspace/dig.py
index db506482..92a0434f 100644
--- a/src/woltspace/dig.py
+++ b/src/woltspace/dig.py
@@ -6,6 +6,7 @@
 import json
 import os
 import re
+import shlex
 import subprocess
 import time
 from contextlib import contextmanager
@@ -65,8 +66,10 @@ def resolve_ssh(
     runner: Callable[..., subprocess.CompletedProcess] = subprocess.run,
 ) -> ResolvedSSH:
     destination = _validate_destination(destination)
+    ssh_config = os.environ.get("WOLTSPACE_DIG_SSH_CONFIG", "").strip()
+    config_args = ["-F", ssh_config] if ssh_config else []
     result = runner(
-        ["ssh", "-G", "--", destination],
+        ["ssh", *config_args, "-G", "--", destination],
         capture_output=True,
         text=True,
         check=False,
@@ -221,12 +224,21 @@ def connect(
         raise DigError("SSH destination no longer resolves to the approved host, user, and port")
     argv = [
         "ssh",
+        *(
+            ["-F", os.environ["WOLTSPACE_DIG_SSH_CONFIG"]]
+            if os.environ.get("WOLTSPACE_DIG_SSH_CONFIG", "").strip()
+            else []
+        ),
         "-o", "StrictHostKeyChecking=yes",
         "--",
         grant["destination"],
     ]
     if command:
-        argv.extend(command)
+        # OpenSSH concatenates every remaining local argv item into one remote
+        # shell command without preserving argument boundaries. Quote once
+        # here so spaces and metacharacters inside an intended argument remain
+        # data when the remote login shell parses it.
+        argv.append(shlex.join(command))
     try:
         result = runner(argv, check=False)
     except KeyboardInterrupt:
diff --git a/test/e2e/dig_ssh_container/Dockerfile b/test/e2e/dig_ssh_container/Dockerfile
new file mode 100644
index 00000000..a5ecf6ec
--- /dev/null
+++ b/test/e2e/dig_ssh_container/Dockerfile
@@ -0,0 +1,26 @@
+FROM python:3.13-slim
+
+RUN apt-get update \
+    && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends openssh-server \
+    && rm -rf /var/lib/apt/lists/* \
+    && useradd --create-home --shell /bin/bash colony \
+    && install -d -m 0755 /run/sshd \
+    && install -d -o colony -g colony -m 0700 /home/colony/.ssh
+
+COPY woltspace-*.whl /tmp/
+COPY authorized_keys /home/colony/.ssh/authorized_keys
+
+RUN python -m pip install --no-cache-dir /tmp/woltspace-*.whl \
+    && rm /tmp/woltspace-*.whl \
+    && chown colony:colony /home/colony/.ssh/authorized_keys \
+    && chmod 0600 /home/colony/.ssh/authorized_keys \
+    && printf '%s\n' \
+        'PasswordAuthentication no' \
+        'KbdInteractiveAuthentication no' \
+        'PermitRootLogin no' \
+        'AllowUsers colony' \
+        >> /etc/ssh/sshd_config \
+    && ssh-keygen -A
+
+EXPOSE 22
+CMD ["/usr/sbin/sshd", "-D", "-e"]
diff --git a/test/e2e/dig_ssh_container/run.py b/test/e2e/dig_ssh_container/run.py
new file mode 100644
index 00000000..0d34ae80
--- /dev/null
+++ b/test/e2e/dig_ssh_container/run.py
@@ -0,0 +1,213 @@
+#!/usr/bin/env python3
+"""Installed-wheel Dig proof against a disposable loopback-only SSH target."""
+
+from __future__ import annotations
+
+import json
+import os
+import secrets
+import shutil
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[3]
+DOCKERFILE = Path(__file__).with_name("Dockerfile")
+
+
+def run(argv, *, env=None, check=True, capture=True, input_text=None):
+    return subprocess.run(
+        [str(item) for item in argv],
+        cwd=ROOT,
+        env=env,
+        check=check,
+        capture_output=capture,
+        text=True,
+        input=input_text,
+    )
+
+
+def wait_for_port(container: str) -> tuple[str, int]:
+    deadline = time.monotonic() + 30
+    while time.monotonic() < deadline:
+        result = run(["docker", "port", container, "22/tcp"], check=False)
+        value = result.stdout.strip()
+        if value:
+            host, port = value.rsplit(":", 1)
+            if host != "127.0.0.1":
+                raise AssertionError(f"SSH was not loopback-only: {value}")
+            return host, int(port)
+        time.sleep(0.2)
+    raise AssertionError("Docker never published the SSH port")
+
+
+def scan_host_key(port: int) -> str:
+    deadline = time.monotonic() + 30
+    while time.monotonic() < deadline:
+        result = run(
+            ["ssh-keyscan", "-T", "2", "-p", str(port), "127.0.0.1"],
+            check=False,
+        )
+        lines = [line for line in result.stdout.splitlines() if line and not line.startswith("#")]
+        if lines:
+            return "\n".join(lines) + "\n"
+        time.sleep(0.2)
+    raise AssertionError("disposable SSH server never offered a host key")
+
+
+def assert_private(path: Path, mode: int) -> None:
+    actual = path.stat().st_mode & 0o777
+    if actual != mode:
+        raise AssertionError(f"{path} mode is {actual:o}, expected {mode:o}")
+
+
+def main() -> int:
+    for command in ("docker", "uv", "ssh", "ssh-keygen", "ssh-keyscan"):
+        if shutil.which(command) is None:
+            raise SystemExit(f"required command is missing: {command}")
+
+    suffix = secrets.token_hex(5)
+    image = f"woltspace-dig-e2e:{suffix}"
+    container = f"woltspace-dig-e2e-{suffix}"
+
+    with tempfile.TemporaryDirectory(prefix="woltspace-dig-e2e-") as raw_temp:
+        temp = Path(raw_temp)
+        context = temp / "context"
+        artifacts = temp / "artifacts"
+        client_home = temp / "client-home"
+        client_wolts = temp / "client-wolts"
+        venv = temp / "client-venv"
+        for directory in (context, artifacts, client_home, client_wolts / "n00b"):
+            directory.mkdir(parents=True)
+
+        key = temp / "dig_key"
+        run(["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", key])
+        shutil.copyfile(f"{key}.pub", context / "authorized_keys")
+
+        run(["uv", "build", "--wheel", "--out-dir", artifacts], capture=False)
+        wheel = next(artifacts.glob("*.whl"))
+        shutil.copyfile(wheel, context / wheel.name)
+
+        try:
+            run(["docker", "build", "-f", DOCKERFILE, "-t", image, context], capture=False)
+            run([
+                "docker", "run", "--rm", "-d", "--name", container,
+                "--publish", "127.0.0.1::22", image,
+            ])
+            _, port = wait_for_port(container)
+
+            known_hosts = temp / "known_hosts"
+            known_hosts.write_text(scan_host_key(port))
+            known_hosts.chmod(0o600)
+            key.chmod(0o600)
+
+            ssh_config = temp / "ssh_config"
+            ssh_config.write_text(
+                "Host dig-e2e\n"
+                "  HostName 127.0.0.1\n"
+                "  User colony\n"
+                f"  Port {port}\n"
+                f"  IdentityFile {key}\n"
+                "  IdentitiesOnly yes\n"
+                "  BatchMode yes\n"
+                f"  UserKnownHostsFile {known_hosts}\n"
+            )
+            ssh_config.chmod(0o600)
+
+            run(["uv", "venv", "--python", "3.13", venv], capture=False)
+            run(["uv", "pip", "install", "--python", venv / "bin/python", wheel], capture=False)
+            cli = venv / "bin/woltspace"
+            env = {
+                **os.environ,
+                "HOME": str(client_home),
+                "WOLTSPACE_WOLTS_DIR": str(client_wolts),
+                "WOLTSPACE_WOLT_NAME": "n00b",
+                "WOLTSPACE_DIG_SSH_CONFIG": str(ssh_config),
+            }
+
+            grant = run([
+                cli, "dig", "grant", "throwaway", "dig-e2e",
+                "--wolt", "n00b", "--json",
+            ], env=env)
+            grant_payload = json.loads(grant.stdout)
+            resolved = grant_payload["grant"]["resolved"]
+            assert resolved == {"hostname": "127.0.0.1", "user": "colony", "port": port}
+
+            whoami = run([
+                cli, "dig", "connect", "--json", "throwaway", "--", "whoami",
+            ], env=env)
+            assert whoami.stdout.strip() == "colony"
+
+            remote_script = (
+                "from pathlib import Path; import json; "
+                "root=Path.home()/'.woltspace'; "
+                "w=root/'wolts'/'seedling'/'wolt'; w.mkdir(parents=True, exist_ok=True); "
+                "(w/'wolt.json').write_text(json.dumps({'name':'seedling','type':'raccoon'})+'\\n'); "
+                "b=root/'bootstrap'; b.mkdir(parents=True, exist_ok=True); "
+                "(b/'dig-handoff.json').write_text(json.dumps({"
+                "'version':'woltspace.dig-handoff/v0','visitor':'n00b',"
+                "'created_colony':'seedling','status':'ready'},sort_keys=True)+'\\n')"
+            )
+            run([
+                cli, "dig", "connect", "--json", "throwaway", "--",
+                "python", "-c", remote_script,
+            ], env=env)
+
+            handoff = run([
+                cli, "dig", "connect", "--json", "throwaway", "--",
+                "cat", ".woltspace/bootstrap/dig-handoff.json",
+            ], env=env)
+            handoff_payload = json.loads(handoff.stdout)
+            assert handoff_payload == {
+                "created_colony": "seedling",
+                "status": "ready",
+                "version": "woltspace.dig-handoff/v0",
+                "visitor": "n00b",
+            }
+
+            remote_version = run([
+                cli, "dig", "connect", "--json", "throwaway", "--",
+                "python", "-c", "import woltspace; print(woltspace.__version__)",
+            ], env=env)
+            assert remote_version.stdout.strip() == "0.5.6"
+
+            listing = json.loads(run([cli, "dig", "list", "--json"], env=env).stdout)
+            audit = listing["grants"][0]
+            assert audit["connect_count"] == 4
+            assert audit["last_exit_code"] == 0
+            store = client_wolts / ".space" / "digs" / "grants.json"
+            assert_private(store.parent, 0o700)
+            assert_private(store, 0o600)
+            assert "PRIVATE KEY" not in store.read_text()
+            assert remote_script not in store.read_text()
+
+            revoked = json.loads(run([
+                cli, "dig", "revoke", "throwaway", "--json",
+            ], env=env).stdout)
+            assert revoked == {"ok": True, "revoked": True, "name": "throwaway"}
+            refused = run([
+                cli, "dig", "connect", "--json", "throwaway", "--", "true",
+            ], env=env, check=False)
+            assert refused.returncode == 1
+            assert json.loads(refused.stdout)["error"] == "unknown dig: throwaway"
+
+            print(json.dumps({
+                "ok": True,
+                "transport": f"127.0.0.1:{port}",
+                "remote_user": "colony",
+                "installed_woltspace": remote_version.stdout.strip(),
+                "handoff": handoff_payload,
+                "connections_audited": audit["connect_count"],
+                "revoked": True,
+            }, indent=2))
+        finally:
+            run(["docker", "rm", "-f", container], check=False)
+            run(["docker", "image", "rm", "-f", image], check=False)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/test/test_dig_cli.py b/test/test_dig_cli.py
index 57bb0b4a..a82c47e1 100644
--- a/test/test_dig_cli.py
+++ b/test/test_dig_cli.py
@@ -33,6 +33,22 @@ def runner(argv, **kwargs):
     assert calls[0][0] == ["ssh", "-G", "--", "my-colony"]
 
 
+def test_isolated_ssh_config_can_be_selected_from_environment(monkeypatch):
+    calls = []
+    monkeypatch.setenv("WOLTSPACE_DIG_SSH_CONFIG", "/tmp/isolated config")
+
+    def runner(argv, **kwargs):
+        calls.append(argv)
+        return subprocess.CompletedProcess(
+            argv, 0, "hostname server.example\nuser colony\nport 22\n", ""
+        )
+
+    resolve_ssh("my-colony", runner=runner)
+    assert calls == [[
+        "ssh", "-F", "/tmp/isolated config", "-G", "--", "my-colony"
+    ]]
+
+
 @pytest.mark.parametrize("destination", ["-oProxyCommand=oops", "host name", "host;id", ""])
 def test_destination_cannot_inject_ssh_options(destination):
     with pytest.raises(DigError):
@@ -70,7 +86,7 @@ def runner(argv, **kwargs):
     assert code == 7
     assert calls == [([
         "ssh", "-o", "StrictHostKeyChecking=yes", "--", "my-colony",
-        "woltspace", "status", "--json",
+        "woltspace status --json",
     ], {"check": False})]
     audited = store.get("new-colony")
     assert audited["connect_count"] == 1
@@ -139,3 +155,22 @@ def test_store_contains_no_credentials_or_command_bodies(tmp_path):
     assert "PRIVATE KEY" not in text
     assert "remote_command" not in text
     assert payload["version"] == "woltspace.digs/v0"
+
+
+def test_remote_command_arguments_are_shell_quoted(tmp_path):
+    store = DigStore(tmp_path / ".space")
+    grant(store)
+    calls = []
+
+    def runner(argv, **kwargs):
+        calls.append(argv)
+        return subprocess.CompletedProcess(argv, 0)
+
+    connect(
+        store,
+        "new-colony",
+        command=("printf", "%s", "hello; not a second command"),
+        resolver=lambda _: resolved(),
+        runner=runner,
+    )
+    assert calls[0][-1] == "printf %s 'hello; not a second command'"