feat(pve): extract agent bridge, client and execution worker

docs/architecture.md

@@ -10,6 +10,9 @@ Notification delivery is separate and never an authorization decision.
 - `privilege_broker.pve.authorization`: immutable requests, proxy identity extraction, signed leases and restricted certificate argv.
 - `privilege_broker.pve.service`: SQLite approval state, real WebAuthn registration/authentication, browser UI and one-time certificate fetch.
 - `privilege_broker.notifications`: server-owned Discord delivery with credential-safe errors and inert agent text.
+- `privilege_broker.pve.client`: mTLS broker HTTP client; the endpoint and trust material are explicit constructor input and transient failures classify separately from final refusals.
+- `privilege_broker.pve.bridge`: agent-side request creation with an ephemeral Ed25519 key, strict response validation and pending-input publication.
+- `privilege_broker.pve.worker`: poll-and-dispatch worker that fetches the lease and the one-time certificate, tries the configured nodes in order and writes terminal result files.
 
 The approval service requires explicitly supplied store, signer and target
 policy. Its environment factory requires private deployment configuration;
@@ -20,8 +23,19 @@ and service isolation are owned by deployment, not by request bodies.
 Browser approval signs a short-lived PVE lease but returns only approval
 status. The bound agent can fetch that lease and a restricted SSH certificate.
 The signer runs only local `ssh-keygen`, not an SSH connection or PVE command.
-No execution worker, target dispatcher or outcome ingestion exists here yet.
-An `approved` request is not an executed operation.
+The bridge and worker exist; the target dispatcher, durable claim store and
+outcome ingestion do not (OpenSpec tasks 4.2 and 5.1). An `approved` request
+is not an executed operation.
+
+The bridge, client and worker take explicit configuration only: they read no
+environment variable and carry no built-in address, path or identity, so a
+missing value fails closed. The worker keeps the broker wire contract
+unchanged: the lease is fetched once, the certificate is fetched once and the
+dispatcher receives `{"action", "ct_id", "lease", "signature"}` on stdin over
+fixed non-interactive SSH options. Terminal results are `succeeded`, `failed`,
+`unknown` (transport timeout; never retried), `error`, `denied`, `cancelled`
+and `expired`. Key and certificate material never appears in results or logs,
+and a consumed certificate is never re-requested.
 
 ## Compatibility and extraction adaptations
 
@@ -60,8 +74,8 @@ delivery is not durable. Product acceptance remains incomplete.
 ## Remaining product stages
 
 The active [OpenSpec change](../openspec/changes/extract-pve-foundation/proposal.md)
-owns authenticated agent lifecycle/outcome interfaces, the bridge, worker,
-dispatcher and end-to-end acceptance. Consumer pinning, deployment and live
+owns the target dispatcher and replay store, the outcome interfaces and
+end-to-end acceptance. Consumer pinning, deployment and live
 PVE acceptance are separate gates. No current consumer has switched.
 
 PVE is the only initial provider. There is no dynamic plugin system, arbitrary
@@ -75,11 +89,14 @@ The extraction uses the owned ai-agents snapshot
 
 | Original file | Product boundary |
 | --- | --- |
-| `MCPs/servers/_pve_common.py` | Policy and canonicalization only; no transport client |
+| `MCPs/servers/_pve_common.py` | Policy, canonicalization and the mTLS broker client; no transport server |
 | `MCPs/servers/pve_ssh_broker.py` | Authorization and certificate argv |
 | `MCPs/servers/pve_ssh_broker_service.py` | Approval store, service, passkey UI and certificate signer |
+| `MCPs/servers/pve_broker_mcp.py` | Agent bridge: request validation, ephemeral key, input publication |
+| `MCPs/servers/pve_broker_worker.py` | Poll-and-dispatch worker: lease/certificate fetch, node failover, result files |
 | `MCPs/servers/_discord_notify.py` | Server-owned notifications, adapted to standalone HTTP and UTC |
 | `MCPs/tests/test_pve_ssh_broker*.py` | Ported synthetic contract tests, supplemented by real local crypto tests |
+| `MCPs/tests/test_pve_broker_mcp.py`, `MCPs/tests/test_pve_broker_worker.py` | Ported synthetic bridge and worker tests |
 
 Private source history, deployment configuration, trust roots, credentials
 and runbooks are not copied. Existing consumers retain their current source

docs/configuration.md

@@ -1,8 +1,9 @@
 # PVE configuration
 
-This pre-release stage supplies authorization, approval and certificate signing,
-not an executor. Deployment, credentials, TLS and consumer cutover remain separate
-operator decisions. No example grants access to a real target.
+This pre-release stage supplies authorization, approval, certificate signing
+and the agent-side bridge and worker, but not the target dispatcher. Deployment,
+credentials, TLS and consumer cutover remain separate operator decisions. No
+example grants access to a real target.
 
 ## Target policy
 
@@ -88,6 +89,68 @@ passkey; denial grants no authority. Protect these URLs against unwanted
 cancellation and metadata disclosure. There is no authenticated generic
 execution-outcome API yet.
 
+## Agent bridge and worker
+
+The bridge and worker are libraries with explicit configuration only. They read
+no environment variable and carry no built-in address, path or identity;
+deployment adapters map their own environment onto these required arguments,
+and a missing value fails closed.
+
+```python
+from pathlib import Path
+from privilege_broker.pve.bridge import ApprovalBridge, BridgeConfig
+from privilege_broker.pve.client import BrokerClient, BrokerClientConfig
+from privilege_broker.pve.worker import WorkerConfig, process_pending_once
+
+client = BrokerClient(
+    BrokerClientConfig(
+        broker_url="https://broker.example.test",  # required https:// URL
+        cert_path="/etc/example/agent-cert.pem",  # required mTLS client certificate
+        key_path="/etc/example/agent-key.pem",  # required mTLS client key
+        ca_path="/etc/example/broker-ca.pem",  # optional trust root
+    )
+)
+bridge = ApprovalBridge(
+    BridgeConfig(
+        state_dir=Path("/var/lib/example-broker"),  # holds input/ results/ keys/
+        public_url="https://broker.example.test",  # approval-page base URL
+        requester="example-agent",  # 1-100 characters
+    )
+)
+worker_config = WorkerConfig(
+    state_dir=Path("/var/lib/example-broker"),
+    targets=("ops@node-a", "ops@node-b"),  # tried in order
+    ssh_port=8922,
+    known_hosts_path="/etc/example/known_hosts",  # pinned host keys
+    poll_interval=5,
+    ssh_timeout=240,
+    expiry_grace=60,
+)
+```
+
+`bridge.request_mutation(action, ct_id, reason, client)` validates the typed
+action, target and reason, writes an ephemeral Ed25519 key (removed again on any
+failure), creates the broker request and publishes
+`input/{request_id}.json`; it returns the `approval_url`. `bridge.read_status`
+returns the worker result, `pending` or `unknown`.
+
+The worker polls `input/` once per `process_pending_once(client, worker_config)`
+call (or forever via `run`). A request past its deadline plus `expiry_grace` is
+expired locally without asking the broker, which bounds all retries. Transient
+broker failures (network, timeout, HTTP 429/502/503/504) before dispatch keep
+the request for the next poll; any final refusal is terminal — in particular
+HTTP 409 after a lost certificate reply ends the request as `error` and never
+requests a second certificate. Dispatch uses fixed non-interactive SSH options
+(`BatchMode`, `IdentitiesOnly`, no TTY, pinned known hosts, configured port) and
+sends `{"action", "ct_id", "lease", "signature"}` on stdin, preserving the
+dispatcher wire contract. A node that does not host the target answers exit 3
+with `not_on_this_node` and the next configured node is tried with the same
+certificate; a transport timeout is `unknown` and is never retried. Results in
+`results/{request_id}.json` are `succeeded`, `failed`, `unknown`, `error`,
+`denied`, `cancelled` or `expired`; key and certificate material never appears
+in results or logs. The target dispatcher and its durable claim store remain
+OpenSpec task 4.2.
+
 ## Certificate source restriction
 
 The library and service require an explicit source restriction:
@@ -106,9 +169,9 @@ The builder only returns argv; the configured service signer invokes local
 `ssh-keygen` with that argv and a timeout. It does not connect to PVE.
 
 A certificate is safe only with the trusted typed dispatcher and lease
-verification on the target. Those execution components are not extracted yet.
-Do not treat certificate issuance as an operation outcome or enable a general
-shell while waiting for the dispatcher.
+verification on the target. The target dispatcher and its claim store are not
+extracted yet (OpenSpec task 4.2). Do not treat certificate issuance as an
+operation outcome or enable a general shell while waiting for the dispatcher.
 
 ## Operator notifications
 

openspec/changes/extract-pve-foundation/tasks.md

@@ -20,7 +20,7 @@
 
 ## 4. PVE execution components
 
-- [ ] 4.1 Extract the agent bridge, client and worker using explicit endpoints, trust and state configuration; verify pre-dispatch transient retry, expiry, lost-certificate terminal failure and no secret-bearing output.
+- [x] 4.1 Extract the agent bridge, client and worker using explicit endpoints, trust and state configuration; verify pre-dispatch transient retry, expiry, lost-certificate terminal failure and no secret-bearing output. [Issue #8](https://github.com/mikomiel/privilege-broker/issues/8).
 - [ ] 4.2 Extract the typed dispatcher and replay store; verify fixed argv, current policy, source/certificate constraints, wrong-node routing, concurrent single claim and restart replay protection.
 - [ ] 4.3 Preserve policy-bounded snapshot pruning and its approval display; verify synthetic old/unrelated/retained snapshot cases and update PVE operation documentation.
 

privilege_broker/pve/__init__.py

@@ -1 +1 @@
-"""PVE policy, canonical authorization and SSH certificate constraints."""
+"""PVE policy, canonical authorization, approval service and the agent-side bridge, client and worker."""

privilege_broker/pve/bridge.py

@@ -0,0 +1,151 @@
+"""Agent-facing approval request bridge.
+
+Creates typed broker requests with an ephemeral Ed25519 key and records the
+pending input file the worker consumes. This is a library: it starts no
+server, reads no environment and carries no deployment default. The approval
+URL it returns points at the operator-facing page of the configured public
+URL.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import uuid
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption, PrivateFormat, PublicFormat
+
+from .client import BrokerClient
+from .common import TYPED_ACTIONS
+
+_UUID_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$")
+
+
+@dataclass(frozen=True)
+class BridgeConfig:
+    """Explicit bridge configuration; no deployment defaults exist."""
+
+    #: Persistent state root; holds ``input/``, ``results/`` and ``keys/``.
+    state_dir: Path
+    #: ``https://`` base URL of the broker's public approval pages.
+    public_url: str
+    #: Requester name sent in every broker request.
+    requester: str
+
+    def __post_init__(self) -> None:
+        if isinstance(self.state_dir, str):
+            object.__setattr__(self, "state_dir", Path(self.state_dir))
+        if not self.public_url.startswith("https://"):
+            raise ValueError("public_url must be an https:// URL")
+        if not isinstance(self.requester, str) or not 1 <= len(self.requester.strip()) <= 100:
+            raise ValueError("requester must be 1-100 characters")
+
+
+class ApprovalBridge:
+    """Creates approval-gated mutation requests and reads their local state.
+
+    Nothing executes here: a request only exists at the broker until a human
+    approves it, and the worker turns an approval into a dispatch.
+    """
+
+    def __init__(self, config: BridgeConfig) -> None:
+        self._config = config
+
+    @property
+    def _input_dir(self) -> Path:
+        return self._config.state_dir / "input"
+
+    @property
+    def _results_dir(self) -> Path:
+        return self._config.state_dir / "results"
+
+    @property
+    def _keys_dir(self) -> Path:
+        return self._config.state_dir / "keys"
+
+    def _validated_create_response(self, response: object) -> tuple[str, int]:
+        """Validate the broker create response before any filesystem publication."""
+        if not isinstance(response, dict):
+            raise ValueError("broker create response must be an object")
+        request_id = response.get("request_id")
+        if not isinstance(request_id, str) or not _UUID_RE.fullmatch(request_id):
+            raise ValueError("broker create response request_id must be a canonical UUID")
+        if response.get("status") != "pending":
+            raise ValueError("broker create response status must be pending")
+        expires_at = response.get("expires_at")
+        if type(expires_at) is not int:
+            raise ValueError("broker create response expires_at must be an integer")
+        return request_id, expires_at
+
+    def request_mutation(self, action: str, ct_id: int, reason: str, client: BrokerClient) -> dict[str, Any]:
+        """Request an approval-gated typed operation; nothing runs until approved.
+
+        Returns ``request_id``, ``status``, ``expires_at`` and the
+        ``approval_url`` to hand to the operator. The ephemeral private key is
+        removed again if the broker call or the response validation fails.
+        """
+        if action not in TYPED_ACTIONS:
+            raise ValueError(f"action must be one of {sorted(TYPED_ACTIONS)}")
+        if type(ct_id) is not int or ct_id < 1:
+            raise ValueError("ct_id must be a positive integer")
+        if not isinstance(reason, str) or not reason.strip() or len(reason) > 500:
+            raise ValueError("reason must be 1-500 characters")
+
+        private_key = Ed25519PrivateKey.generate()
+        key_id = uuid.uuid4().hex
+        self._keys_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
+        key_path = self._keys_dir / f"{key_id}.key"
+        key_path.write_bytes(private_key.private_bytes(Encoding.PEM, PrivateFormat.OpenSSH, NoEncryption()))
+        key_path.chmod(0o600)
+        public_key = private_key.public_key().public_bytes(Encoding.OpenSSH, PublicFormat.OpenSSH).decode("ascii")
+
+        try:
+            response = client.request(
+                "POST",
+                "/v1/broker/requests",
+                {
+                    "action": action,
+                    "ct_id": ct_id,
+                    "reason": reason,
+                    "requester": self._config.requester,
+                    "ssh_public_key": public_key,
+                },
+            )
+            request_id, expires_at = self._validated_create_response(response)
+        except Exception:
+            # The ephemeral private key never survives a failed request.
+            key_path.unlink(missing_ok=True)
+            raise
+
+        self._input_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
+        input_data = {
+            "request_id": request_id,
+            "action": action,
+            "ct_id": ct_id,
+            "reason": reason,
+            "public_key": public_key,
+            "key_id": key_id,
+            "expires_at": expires_at,
+        }
+        (self._input_dir / f"{request_id}.json").write_text(json.dumps(input_data), encoding="utf-8")
+        return {
+            "request_id": request_id,
+            "status": "pending",
+            "expires_at": expires_at,
+            "approval_url": f"{self._config.public_url.rstrip('/')}/v1/broker/requests/{request_id}/approval",
+        }
+
+    def read_status(self, request_id: str) -> dict[str, Any]:
+        """Return the worker's terminal result, ``pending``, or ``unknown``."""
+        if not isinstance(request_id, str) or not _UUID_RE.fullmatch(request_id):
+            raise ValueError("request_id must be a canonical UUID")
+        result_file = self._results_dir / f"{request_id}.json"
+        if result_file.exists():
+            return json.loads(result_file.read_text(encoding="utf-8"))
+        if (self._input_dir / f"{request_id}.json").exists():
+            return {"status": "pending"}
+        return {"status": "unknown", "error": "request not found or expired"}