"""HAVEN Core Python client: local keys, DPoP sessions and explicit capabilities."""

import base64
import hashlib
import json
import secrets
import time
import uuid
from urllib.parse import quote, unquote, urlsplit

import httpx
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec, ed25519
from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature


def canonical(value):
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False)


def b64(value):
    return base64.urlsafe_b64encode(value).rstrip(b"=").decode()


def unb64(value):
    raw = base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
    if b64(raw) != value:
        raise ValueError("Expected canonical base64url")
    return raw


def digest(value):
    return hashlib.sha256(value).hexdigest()


def component(value):
    return quote(value, safe="")


def _explicit_v4_capability(value):
    if not isinstance(value, str) or not value.strip() or len(value) > 100:
        raise ValueError("An explicit exact-operation V4 capability is required")


def _v4_page(offset):
    if type(offset) is not int or not 0 <= offset <= 10000:
        raise ValueError("V4 offset must be an integer between 0 and 10000")
    return "?offset=" + str(offset) if offset else ""


def _product_capability(value):
    if not isinstance(value, str) or not value.strip() or len(value) > 100:
        raise ValueError("An explicit exact-operation capability is required")


def _product_integer(value, name, minimum=1, maximum=2**53 - 1):
    if type(value) is not int or not minimum <= value <= maximum:
        raise ValueError(name + " must be a bounded safe integer")
    return value


def _product_boolean(value):
    if type(value) is not bool:
        raise ValueError("Use a boolean read state")
    return value


class HavenError(Exception):
    def __init__(self, code, message, status=0, request_id=None, details=None, uncertain=False):
        super().__init__(f"{code}: {message}")
        self.code, self.status, self.request_id = code, status, request_id
        self.details, self.uncertain = details or {}, uncertain


class HavenIdentity:
    def __init__(self, key):
        self._key = key
        self.public_key = {"kty": "OKP", "crv": "Ed25519", "x": b64(key.public_key().public_bytes_raw())}
        self.key_identity = "agent:" + b64(hashlib.sha256(canonical(self.public_key).encode()).digest())

    @classmethod
    def generate(cls):
        return cls(ed25519.Ed25519PrivateKey.generate())

    @classmethod
    def from_key(cls, key):
        raw = unb64(key) if isinstance(key, str) else bytes(key)
        return cls(ed25519.Ed25519PrivateKey.from_private_bytes(raw))

    def export_key(self):
        """Explicit caller-controlled backup; never automatically persisted or sent."""
        return self._key.private_bytes_raw()

    def sign(self, message):
        return b64(self._key.sign(message))

    def __repr__(self):
        return f"HavenIdentity({self.key_identity})"


class HavenClient:
    def __init__(self, origin, *, allow_local=False, timeout=15):
        parts = urlsplit(origin)
        local = parts.hostname in {"localhost", "127.0.0.1", "::1"}
        if (
            parts.scheme not in {"http", "https"}
            or not parts.hostname
            or parts.username
            or parts.password
            or parts.path not in {"", "/"}
            or parts.query
            or parts.fragment
            or parts.scheme == "http"
            and not (allow_local and local)
        ):
            raise ValueError("Use HTTPS node origin; literal loopback HTTP requires allow_local=True")
        self.origin = origin.rstrip("/")
        self._http = httpx.Client(timeout=timeout, trust_env=False, follow_redirects=False)

    def _request(self, method, path, body=None, session=None, capability=None, *, skip_legacy=False):
        if not path.startswith("/") or path.startswith("//") or "#" in path:
            raise ValueError("Use a node-relative API path")
        method = method.upper()
        generation = session._generation if session else None
        if session and not capability and not skip_legacy:
            capability = session._legacy_capability(method, path)
            session._guard(generation)
        raw = canonical(body).encode() if body is not None else b""
        headers = {"Accept": "application/json"}
        if body is not None:
            headers["Content-Type"] = "application/json"
        if session:
            headers.update(session._headers(method, path, raw))
            session._guard(generation)
        if capability:
            headers["X-Haven-Capability"] = capability
        try:
            response = self._http.request(method, self.origin + path, content=raw, headers=headers)
        except httpx.HTTPError:
            raise HavenError(
                "TRANSPORT_UNAVAILABLE",
                "Request outcome is unknown; inspect state before retrying.",
                uncertain=method not in {"GET", "HEAD", "OPTIONS"},
            ) from None
        if 300 <= response.status_code < 400:
            raise HavenError("REDIRECT_REJECTED", "Node redirects are not followed.", response.status_code)
        try:
            value = response.json()
        except ValueError:
            raise HavenError("INVALID_RESPONSE", "Expected a JSON response.", response.status_code) from None
        if response.is_error:
            error = value.get("error", {}) if isinstance(value, dict) else {}
            raise HavenError(
                error.get("code", "HTTP_ERROR"),
                error.get("message", "Node rejected request."),
                response.status_code,
                error.get("request_id") or response.headers.get("X-Request-ID"),
                error.get("details"),
            )
        if session:
            session._guard(generation)
        return value

    def discover(self):
        return self._request("GET", "/.well-known/haven")

    def federation_v4_manifest(self):
        """Read the public V4 manifest; reading does not pin or admit a peer."""
        return self._request("GET", "/api/v1/federation/v4/tasks/manifest")

    def identity_status(self, key_identity):
        """Read minimal public key state without a session or recovery secret."""
        return self._request("GET", f"/api/v1/identity/status/{component(key_identity)}")

    def connect(self, identity, *, mode="GENESIS", runtime_id="python-sdk", device_name=""):
        dpop = ec.generate_private_key(ec.SECP256R1())
        point = dpop.public_key().public_numbers()
        public = {"kty": "EC", "crv": "P-256", "x": b64(point.x.to_bytes(32)), "y": b64(point.y.to_bytes(32))}
        challenge = self._request(
            "POST",
            "/api/v1/identity/challenge",
            {
                "publicKey": identity.public_key,
                "dpopPublicKey": public,
                "arrivalMode": mode,
            },
        )
        expected = (
            b"HAVEN_DPOP_BINDING_V1\0"
            + unb64(challenge["challenge"])
            + b"\0"
            + b64(hashlib.sha256(canonical(public).encode()).digest()).encode()
        )
        if challenge["agentId"] != identity.key_identity or unb64(challenge["proofMessage"]) != expected:
            raise HavenError(
                "CHALLENGE_BINDING_INVALID", "Node challenge is not bound to this identity and proof key."
            )
        verified = self._request(
            "POST",
            "/api/v1/identity/verify",
            {
                "challengeId": challenge["challengeId"],
                "publicKey": identity.public_key,
                "signature": identity.sign(unb64(challenge["proofMessage"])),
            },
        )
        if verified["agentId"] != identity.key_identity:
            raise HavenError("IDENTITY_MISMATCH", "Node returned another key identity.")
        data = self._request(
            "POST",
            "/api/v1/session",
            {
                "verificationTicket": verified["verificationTicket"],
                "runtimeId": runtime_id,
                "deviceName": device_name,
            },
        )
        session = HavenSession(self, identity, dpop, public, data)
        session.principal_id = session.request("GET", "/api/v1/principals/me")["id"]
        return session

    def close(self):
        self._http.close()

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.close()


class HavenSession:
    def __init__(self, client, identity, dpop, public, data):
        self.client, self.identity, self._dpop, self._dpop_public = client, identity, dpop, public
        self.principal_id = None
        self._generation = 0
        self._legacy_contract = None
        self._legacy_capabilities = {}
        self._replace_token(data)

    def _replace_token(self, data):
        self._generation += 1
        self._legacy_contract = None
        self._legacy_capabilities.clear()
        self._token = data["accessToken"]
        self.metadata = {key: value for key, value in data.items() if key != "accessToken"}

    def _guard(self, generation):
        if generation != self._generation or not self._token:
            raise HavenError("SESSION_CHANGED", "The session changed; inspect saved state before retrying.")

    def _legacy_capability(self, method, path):
        generation = self._generation
        if self._legacy_contract is None:
            contract = self.client._request("GET", "/api/v1/policy/legacy-contract")
            self._guard(generation)
            self._legacy_contract = contract
        self._guard(generation)
        contract = self._legacy_contract
        if (
            contract.get("action") != "legacy.invoke"
            or not isinstance(contract.get("requireCapabilities"), bool)
            or not isinstance(contract.get("exemptPaths"), list)
            or not isinstance(contract.get("exemptPrefixes"), list)
        ):
            raise HavenError("LEGACY_CONTRACT_INVALID", "Unsupported legacy invocation policy contract.")
        raw_path = self.client._http.build_request(method, self.client.origin + path).url.raw_path.split(
            b"?"
        )[0]
        decoded = unquote(raw_path.decode("ascii"))
        if (
            not contract["requireCapabilities"]
            or decoded in contract["exemptPaths"]
            or any(
                decoded == prefix or decoded.startswith(prefix + "/") for prefix in contract["exemptPrefixes"]
            )
        ):
            return None
        resource = "legacy:" + method + ":" + hashlib.sha256(raw_path).hexdigest()
        cached = self._legacy_capabilities.get(resource)
        if cached and cached["expires_at"] > time.time() + 5:
            return cached["id"]
        value = self.client._request(
            "POST",
            "/api/v1/capabilities/request",
            {
                "action": "legacy.invoke",
                "resource": resource,
                "purpose": "User-selected exact SDK operation",
                "ttlSeconds": 600,
            },
            self,
            skip_legacy=True,
        )
        self._guard(generation)
        self._legacy_capabilities[resource] = value
        return value["id"]

    def _headers(self, method, path, raw):
        if not self._token:
            raise HavenError("SESSION_CLOSED", "Reconnect with your identity key.")
        header = b64(canonical({"typ": "dpop+jwt", "alg": "ES256", "jwk": self._dpop_public}).encode())
        claims = {
            "jti": secrets.token_urlsafe(24),
            "iat": int(time.time()),
            "htm": method,
            "htu": self.client.origin + path.split("?")[0],
            "ath": b64(hashlib.sha256(self._token.encode()).digest()),
            "bht": b64(hashlib.sha256(raw).digest()),
        }
        part = header + "." + b64(canonical(claims).encode())
        r, s = decode_dss_signature(self._dpop.sign(part.encode(), ec.ECDSA(hashes.SHA256())))
        return {
            "Authorization": "DPoP " + self._token,
            "DPoP": part + "." + b64(r.to_bytes(32) + s.to_bytes(32)),
        }

    def request(self, method, path, body=None, *, capability=None):
        return self.client._request(method, path, body, self, capability)

    def request_capability(self, action, resource, purpose, **constraints):
        return self.request(
            "POST",
            "/api/v1/capabilities/request",
            {
                "action": action,
                "resource": resource,
                "purpose": purpose,
                **constraints,
            },
        )

    def change_identity_status(self, action, version, reason, capability, *, expires_at=None):
        """Sign a V1 status change locally. Requires an explicit identity.manage capability."""
        if action not in {"set-expiry", "expire", "compromise"}:
            raise ValueError("Unsupported identity status action")
        body = {"action": action, "keyIdentity": self.identity.key_identity, "version": version,
                "expiresAt": expires_at, "issuedAt": int(time.time()),
                "nonce": secrets.token_urlsafe(24), "reason": reason}
        body["signature"] = self.identity.sign(b"HAVEN_IDENTITY_STATUS_V1\0" + canonical(body).encode())
        result = self.request("POST", "/api/v1/identity/status", body, capability=capability)
        if action in {"expire", "compromise"}:
            self._token = None
            self._generation += 1
            self._legacy_contract = None
            self._legacy_capabilities.clear()
        return result

    def delegate(self, capability, subject, purpose, **constraints):
        return self.request(
            "POST",
            f"/api/v1/capabilities/{component(capability)}/delegate",
            {
                "subject": subject,
                "purpose": purpose,
                **constraints,
            },
        )

    def revoke_capability(self, capability):
        return self.request("POST", f"/api/v1/capabilities/{component(capability)}/revoke")

    def renew(self):
        """Explicit single token rotation. Never retry an uncertain rotation."""
        result = self.request("POST", "/api/v1/session/renew")
        self._replace_token(result)
        return self.metadata.copy()

    def disconnect(self):
        try:
            return self.request("POST", "/api/v1/session/close")
        finally:
            self._token = None
            self._generation += 1
            self._legacy_contract = None
            self._legacy_capabilities.clear()

    def memory_write(
        self,
        object_id,
        payload,
        capability,
        *,
        namespace="/private",
        version=0,
        event_type="CREATE",
        acl=None,
        source="python-sdk",
        provenance=None,
    ):
        return self.request(
            "POST",
            "/api/v1/memory/events",
            {
                "objectId": object_id,
                "payload": payload,
                "namespace": namespace,
                "version": version,
                "eventType": event_type,
                "acl": acl or [],
                "source": source,
                "provenance": provenance or {},
            },
            capability=capability,
        )

    def memory_read(self, capability, *, owner=None, namespace="/private", query="", offset=0, search_mode=None):
        if search_mode is not None and search_mode not in {"lexical", "semantic", "hybrid"}:
            raise ValueError("Use lexical, semantic or hybrid memory search")
        return self.request(
            "POST",
            "/api/v1/memory/retrieve",
            {
                "owner": owner or self.principal_id,
                "namespace": namespace,
                "query": query,
                "offset": offset,
                **({"searchMode": search_mode} if search_mode is not None else {}),
            },
            capability=capability,
        )

    def restore(self, capability, memory_capabilities=(), *, task_capabilities=(), record_snapshot=False, cursor=None):
        return self.request(
            "POST",
            "/api/v1/agents/continuity/restore",
            {
                "memoryCapabilities": list(memory_capabilities),
                "taskCapabilities": list(task_capabilities),
                "recordSnapshot": record_snapshot,
                **({"cursor": cursor} if cursor is not None else {}),
            },
            capability=capability,
        )

    def principal_create(self, kind, name, capability, *, description="", collective_id=None):
        if kind not in {"HUMAN", "ORG", "ORGANIZATION", "SERVICE"}:
            raise ValueError("Managed principals use HUMAN, ORG, ORGANIZATION or SERVICE")
        return self.request("POST", "/api/v1/principals", {
            "type": kind, "name": name, "description": description,
            **({"collectiveId": collective_id} if collective_id is not None else {}),
        }, capability=capability)

    def principal_read(self, identifier, capability):
        return self.request("GET", f"/api/v1/principals/{component(identifier)}", capability=capability)

    def principal_list(self, capability, *, offset=0):
        return self.request("GET", f"/api/v1/principals?offset={offset}", capability=capability)

    def portable_execution_export(self, artifact, capability):
        return self.request("GET", f"/api/v1/proofs/{component(artifact)}/portable-v2", capability=capability)

    def portable_execution_verify(self, proof, capability, *, expected_node_id, expected_module_hash, expected_input_hash):
        return self.request("POST", "/api/v1/proofs/verify-portable-v2", {
            "proof": proof, "expectedNodeId": expected_node_id,
            "expectedModuleHash": expected_module_hash, "expectedInputHash": expected_input_hash,
        }, capability=capability)

    def _federation_v4_request(self, method, path, capability, body=None):
        # Empty capabilities must never fall through to legacy auto-acquisition.
        _explicit_v4_capability(capability)
        return self.client._request(method, "/api/v1/federation/v4/tasks" + path,
                                    body, self, capability, skip_legacy=True)

    def federation_v4_workspace(self, capability, *, offset=0):
        """Read only this participant's V4 inventory with task.list:<actor>."""
        return self._federation_v4_request("GET", "/workspace" + _v4_page(offset), capability)

    def federation_v4_exchange(self, task_id, capability, *, offset=0, offer_id=None):
        query = _v4_page(offset)
        if offer_id is not None:
            if not isinstance(offer_id, str) or not 1 <= len(offer_id) <= 100:
                raise ValueError("V4 offer ID must contain 1 to 100 characters")
            query += ("&" if query else "?") + "offerId=" + component(offer_id)
        return self._federation_v4_request("GET", f"/{component(task_id)}/exchange" + query, capability)

    def federation_v4_offer(self, task_id, peer_id, runtime, capability, *, ttl_seconds=3600):
        """Explicitly send an offer, without transferring memory or starting work."""
        return self._federation_v4_request("POST", f"/{component(task_id)}/offer", capability, {
            "peerId": peer_id, "ttlSeconds": ttl_seconds, "runtime": runtime,
        })

    def federation_v4_received(self, capability, *, offset=0):
        return self._federation_v4_request("GET", "/received" + _v4_page(offset), capability)

    def federation_v4_accept(self, offer_id, resource_id, expected_resource_version,
                             runtime_capability, capability):
        """Map the exact approved resource; accepting does not execute it."""
        if not isinstance(runtime_capability, str) or not runtime_capability.strip() or len(runtime_capability) > 100:
            raise ValueError("An explicit runtime.create capability is required")
        return self._federation_v4_request("POST", f"/received/{component(offer_id)}/accept", capability, {
            "expectedState": "OFFERED", "resourceId": resource_id,
            "expectedResourceVersion": expected_resource_version, "runtimeCapability": runtime_capability,
        })

    def federation_v4_send_result(self, task_id, offer_id, capability):
        """Explicitly return the already accepted local execution proof."""
        return self._federation_v4_request("POST", f"/{component(task_id)}/result", capability, {
            "offerId": offer_id,
        })

    def federation_v4_results(self, task_id, capability):
        return self._federation_v4_request("GET", f"/{component(task_id)}/results", capability)

    def federation_v4_review(self, task_id, result_id, decision, reason, task_version,
                             capability, *, issued_at=None):
        """Sign this creator's exact V4 decision; never accept automatically."""
        _explicit_v4_capability(capability)
        generation = self._generation
        self._guard(generation)
        if decision not in {"ACCEPTED", "REJECTED"}:
            raise ValueError("Use ACCEPTED or REJECTED")
        issued_at = int(time.time()) if issued_at is None else issued_at
        if (type(task_version) is not int or not 1 <= task_version <= 2**53 - 1
                or type(issued_at) is not int or not 0 <= issued_at <= 2**53 - 1):
            raise ValueError("Task version and issued time must be safe integers")
        if not isinstance(reason, str) or not 1 <= len(reason) <= 2000 or not self.principal_id:
            raise ValueError("A review reason and connected principal are required")
        signed = {"source_task": task_id, "result_id": result_id, "actor": self.principal_id,
                  "decision": decision, "reason": reason, "task_version": task_version, "issued_at": issued_at}
        signature = self.identity.sign(b"HAVEN_FEDERATION_DECISION_V4\0" + canonical(signed).encode())
        self._guard(generation)
        return self._federation_v4_request("POST", f"/{component(task_id)}/review", capability, {
            "resultId": result_id, "decision": decision, "reason": reason,
            "taskVersion": task_version, "issuedAt": issued_at, "signature": signature,
        })

    def task_create(self, requirements, capability, **options):
        return self.request(
            "POST", "/api/v1/tasks", {"requirements": requirements, **options}, capability=capability
        )

    def task_read(self, task, capability):
        return self.request("GET", f"/api/v1/tasks/{component(task)}", capability=capability)

    def task_transition(self, task, version, status, capability):
        return self.request(
            "POST",
            f"/api/v1/tasks/{component(task)}/transition",
            {
                "version": version,
                "status": status,
            },
            capability=capability,
        )

    def task_claim(self, task, version, capability):
        return self.request(
            "POST", f"/api/v1/tasks/{component(task)}/claim", {"version": version}, capability=capability
        )

    def task_start(self, task, version, capability, **options):
        return self.request(
            "POST",
            f"/api/v1/tasks/{component(task)}/runs",
            {
                "version": version,
                **options,
            },
            capability=capability,
        )

    def artifact_upload(
        self, body, capability, *, mime_type="text/plain", task=None, run=None, visibility="private"
    ):
        checksum = digest(body.encode())
        signed = {
            "sha256": checksum,
            "mime_type": mime_type,
            "task_id": task,
            "run_id": run,
            "author": self.principal_id,
        }
        signature = self.identity.sign(b"HAVEN_ARTIFACT_V1\0" + canonical(signed).encode())
        return self.request(
            "POST",
            "/api/v1/artifacts",
            {
                "body": body,
                "mimeType": mime_type,
                "taskId": task,
                "runId": run,
                "visibility": visibility,
                "signature": signature,
            },
            capability=capability,
        )

    def artifact_read(self, artifact, capability):
        return self.request("GET", f"/api/v1/artifacts/{component(artifact)}", capability=capability)

    def task_submit(self, task, run, artifact, version, capability):
        return self.request(
            "POST",
            f"/api/v1/tasks/{component(task)}/runs/{component(run)}/submit",
            {
                "version": version,
                "artifactId": artifact,
            },
            capability=capability,
        )

    def verify_artifact(
        self, artifact, checksum, capability, *, method="sha256", result="VALID", evidence=""
    ):
        created = int(time.time())
        signed = {
            "artifact_id": artifact,
            "sha256": checksum,
            "verifier": self.principal_id,
            "method": method,
            "result": result,
            "evidence": evidence,
            "created_at": created,
        }
        return self.request(
            "POST",
            f"/api/v1/artifacts/{component(artifact)}/verifications",
            {
                "method": method,
                "result": result,
                "evidence": evidence,
                "createdAt": created,
                "signature": self.identity.sign(b"HAVEN_VERIFICATION_V1\0" + canonical(signed).encode()),
            },
            capability=capability,
        )

    def proof(self, artifact, capability):
        return self.request("GET", f"/api/v1/proofs/{component(artifact)}", capability=capability)

    def message_send(
        self, recipient, body, capability, *, message_type="message", correlation_id=None, ttl=300
    ):
        created = int(time.time())
        envelope = {
            "id": str(uuid.uuid4()),
            "sender": self.principal_id,
            "recipient": recipient,
            "type": message_type,
            "created_at": created,
            "expires_at": created + ttl,
            "body": body,
            "correlation_id": correlation_id or str(uuid.uuid4()),
        }
        return self.request(
            "POST",
            "/api/v1/a2a/messages",
            {
                **envelope,
                "signature": self.identity.sign(b"HAVEN_A2A_MESSAGE_V1\0" + canonical(envelope).encode()),
            },
            capability=capability,
        )

    def messages(self, capability, *, direction="inbox", offset=0):
        return self.request(
            "GET",
            "/api/v1/a2a/messages?direction=" + component(direction) + "&offset=" + str(offset),
            capability=capability,
        )

    def _product_request(self, method, path, capability, body=None):
        _product_capability(capability)
        return self.client._request(method, path, body, self, capability, skip_legacy=True)

    def notifications(self, capability, *, offset=0, unread_only=False):
        _product_integer(offset, "offset", 0, 10000)
        _product_boolean(unread_only)
        return self._product_request("GET", "/api/v1/notifications?offset=" + str(offset)
                                     + "&unreadOnly=" + str(unread_only).lower(), capability)

    def notification_unread(self, capability):
        return self._product_request("GET", "/api/v1/notifications/unread", capability)

    def notification_mark_read(self, identifier, read, version, capability):
        return self._product_request("PUT", f"/api/v1/notifications/{component(identifier)}/read", capability, {
            "read": _product_boolean(read), "version": _product_integer(version, "version"),
        })

    def message_mark_read(self, identifier, read, capability):
        return self._product_request("PUT", f"/api/v1/inbox/{component(identifier)}/read", capability, {
            "read": _product_boolean(read),
        })

    def improvements(self, capability, *, offset=0):
        _product_integer(offset, "offset", 0, 10000)
        return self._product_request("GET", "/api/v1/improvements?offset=" + str(offset), capability)

    def improvement_read(self, identifier, capability):
        return self._product_request("GET", f"/api/v1/improvements/{component(identifier)}", capability)

    def improvement_create(self, program_id, objective, cases, capability, *, max_iterations,
                           total_token_budget, deadline_seconds):
        """Create an inert experiment with explicit limits; does not call a provider."""
        return self._product_request("POST", "/api/v1/improvements", capability, {
            "programId": program_id, "objective": objective, "cases": cases,
            "maxIterations": _product_integer(max_iterations, "maxIterations"),
            "totalTokenBudget": _product_integer(total_token_budget, "totalTokenBudget"),
            "deadlineSeconds": _product_integer(deadline_seconds, "deadlineSeconds"),
        })

    def improvement_candidate(self, identifier, version, instructions, capability):
        return self._product_request("POST", f"/api/v1/improvements/{component(identifier)}/candidates", capability, {
            "version": _product_integer(version, "version"), "instructions": instructions,
        })

    def improvement_start(self, identifier, version, secret_capability, capability):
        """Explicit bounded provider consent; a separate secret.use capability is required."""
        _product_capability(secret_capability)
        return self._product_request("POST", f"/api/v1/improvements/{component(identifier)}/start", capability, {
            "version": _product_integer(version, "version"), "secretCapability": secret_capability,
        })

    def improvement_cancel(self, identifier, version, capability):
        return self._product_request("POST", f"/api/v1/improvements/{component(identifier)}/cancel", capability, {
            "version": _product_integer(version, "version"),
        })

    def improvement_promote(self, identifier, version, program_version, reason, capability):
        return self._product_request("POST", f"/api/v1/improvements/{component(identifier)}/promote", capability, {
            "version": _product_integer(version, "version"),
            "programVersion": _product_integer(program_version, "programVersion"), "reason": reason,
        })

    def improvement_rollback(self, identifier, version, program_version, reason, capability):
        return self._product_request("POST", f"/api/v1/improvements/{component(identifier)}/rollback", capability, {
            "version": _product_integer(version, "version"),
            "programVersion": _product_integer(program_version, "programVersion"), "reason": reason,
        })

    def __repr__(self):
        return f"HavenSession(principal_id={self.principal_id!r})"


def verify_artifact_locally(artifact):
    """Authorship and byte integrity only; does not establish truth or execution."""
    if digest(artifact["body"].encode()) != artifact["sha256"]:
        return False
    signed = {
        "sha256": artifact["sha256"],
        "mime_type": artifact["mime_type"],
        "task_id": artifact["task_id"],
        "run_id": artifact["run_id"],
        "author": artifact["author"],
    }
    key = artifact["public_key"]
    if key.get("kty") != "OKP" or key.get("crv") != "Ed25519":
        return False
    try:
        ed25519.Ed25519PublicKey.from_public_bytes(unb64(key["x"])).verify(
            unb64(artifact["signature"]), b"HAVEN_ARTIFACT_V1\0" + canonical(signed).encode()
        )
    except (InvalidSignature, ValueError):
        return False
    return True