Scope boundary

Owns containment of desktop GUI and OS-level computer-use agents in disposable logged-out workspaces, privileged UI observation, screenshot-risk handling, OS automation allowlists, stable accessibility invocation, action-bound confirmation consumption, signed UI or API effect readback, watch-mode halt, and deterministic cleanup. Workspace/platform and agent-harness components implement one desktop containment outcome through the signed launch, observation, approval and cleanup interfaces. Verification covers the whole applicable workspace/action population and current supervision evidence.

AID-I-008.001: AID-I-008.004 isolates a complete desktop or VDI computer-use workspace; AID-I-008.001 isolates browser contexts and browser-managed storage inside a browser runtime. Desktop files, applications, password managers, device APIs, and OS session residue require this broader workspace boundary.

AID-I-008.002: AID-I-008.004 applies privileged desktop UI observation and invocation; AID-I-008.002 intercepts and binds sensitive browser requests and cross-origin writes. A desktop click may not correspond to a browser request, while browser request suspension does not constrain other OS applications or device surfaces.

AID-I-008.003: AID-I-008.004 contains the desktop workspace and its general clipboard boundary; AID-I-008.003 owns browser download and clipboard transfer quarantine and clean-context handoff. Use the transfer control for moving bytes across the browser boundary and this control for the wider OS workspace and action surface.

AID-I-008.005: AID-I-008.004 contains general desktop and UI automation; AID-I-008.005 brokers authentication-bearing browser navigation in a dedicated context. A disposable desktop does not authorize exposing raw magic links, codes, tokens, or callback state to the model.

AID-H-018.003: AID-I-008.004 binds and invokes the exact desktop UI action and reads back its effect; AID-H-018.003 independently validates the high-impact action and issues the canonical approval receipt. The desktop adapter consumes that receipt and must not become a second approval authority.

AID-H-018.006: AID-I-008.004 adds desktop-specific re-observation and effect evidence; AID-H-018.006 owns generic execution-time authorization re-verification and anti-TOCTOU policy. Both checks remain necessary when screen state, target identity, or effective authority can change between approval and invocation.

AID-H-019: AID-I-008.004 owns OS and desktop workspace containment; AID-H-019 owns URL admission, safe fetch, browser-content demotion, and sanitized observation export. Safe web-content handling does not constrain local apps, files, screenshots, or accessibility actions outside the browser surface.

Warning

Desktop Automation Has Host-Level Blast Radius

Desktop computer-use agents can click, type, read screenshots, move files, and interact with applications that were never designed for automated agents. Treat the workspace as a high-blast-radius environment: start from a logged-out disposable profile, grant only task-required applications and directories, stop on visual prompt-injection signals, and require approval before irreversible actions.

Implementation guidance

• AID-I-008.004-G001: Deploy and attest a peer-bound privileged OS/accessibility observer with a durable signed evidence service for disposable desktop workspaces.
• AID-I-008.004-G002: Run computer-use agents in disposable, logged-out desktop workspaces with explicit app, file, network, and clipboard boundaries.
• AID-I-008.004-G003: Gate every high-risk screenshot population through a signed independent classifier or human decision with non-overwrite evidence readback.
• AID-I-008.004-G004: Consume the canonical H-018 receipt for one exact privileged desktop UI binding, invoke only that stable accessibility element, and replay the signed post-action UI/API effect evidence before releasing the action result.
• AID-I-008.004-G005: Enforce signed-policy watch mode with authenticated kill-switch receipts, durable signed evidence, and deterministic workspace cleanup.

AID-I-008.004-G001

Required platform service and inputs

Deploy this service on Linux disposable desktop workspaces that expose a dedicated AT-SPI accessibility bus to a privileged observer outside the model workspace. Require the signed workspace launch receipt produced by the disposable-workspace launcher in AID-I-008.004-G002 and the signed target/effect policy consumed by AID-H-018.003. Missing, expired, unverified, or context-mismatched launch or policy receipts are INSUFFICIENT_DATA. The service authenticates the trusted desktop dispatcher peer, derives the current application, window, and element from OS accessibility APIs, holds one lock lease across activate, re-observe, invoke, readback, and freeze operations, and durably stores and reads back a signed result receipt. The desktop-action executor combines these observations with a consumed H-018 receipt and must re-verify both before authorization.

Root-owned service contract

Install PyGObject/AT-SPI, PyNaCl, and Requests from pinned packages. Run the service under a dedicated host service account with access to the task's dedicated AT-SPI bus, the root-owned workspace registry, H-018 verification key, evidence signing key, and workspace-control mTLS identity. Expose the Unix socket only to the separately sandboxed trusted dispatcher UID/cgroup recorded by the launcher; never mount it into the model workspace. Registry and policy envelopes use detached Ed25519 signatures over RFC 8785/JCS canonical JSON and are written by the privileged launcher.

# File: computer_use/desktop_observer_service.py
from __future__ import annotations

import base64
import fcntl
import hashlib
import json
import math
import os
import re
import socket
import sqlite3
import stat
import struct
import time
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Sequence
from uuid import uuid4

import gi
gi.require_version("Atspi", "2.0")
from gi.repository import Atspi
import jcs
import requests
from nacl.exceptions import BadSignatureError
from nacl.signing import SigningKey, VerifyKey
from computer_use.desktop_watch_readiness import require_watch_readback

SOCKET_PATH = "/run/aidefend/desktop-observer.sock"
SERVICE_IDENTITY = "aidefend-desktop-observer-prod"
REGISTRY_DIR = Path("/run/aidefend/desktop-workspaces")
EVIDENCE_DB = Path("/var/lib/aidefend/desktop-observer/evidence.sqlite3")
MAX_FRAME = 1_048_576
ID = re.compile(r"[A-Za-z0-9._:-]{1,128}")
SHA256 = re.compile(r"sha256:[0-9a-f]{64}")
OUTCOMES = {"PASS", "FAIL", "INSUFFICIENT_DATA", "NOT_APPLICABLE", "ERROR"}
REGISTRY_VERIFY_KEY = VerifyKey(bytes.fromhex(os.environ["WORKSPACE_REGISTRY_VERIFY_KEY_HEX"]))
POLICY_VERIFY_KEY = VerifyKey(bytes.fromhex(os.environ["DESKTOP_POLICY_VERIFY_KEY_HEX"]))
H018_VERIFY_KEY = VerifyKey(bytes.fromhex(os.environ["H018_RECEIPT_VERIFY_KEY_HEX"]))
READBACK_VERIFY_KEY = VerifyKey(bytes.fromhex(os.environ["DESKTOP_READBACK_VERIFY_KEY_HEX"]))
CONTROL_VERIFY_KEY = VerifyKey(bytes.fromhex(os.environ["WORKSPACE_CONTROL_VERIFY_KEY_HEX"]))
EVIDENCE_KEY = SigningKey(bytes.fromhex(os.environ["DESKTOP_EVIDENCE_SIGNING_KEY_HEX"]))
CONTROL_URL = os.environ["WORKSPACE_CONTROL_URL"].rstrip("/")
if not CONTROL_URL.startswith("https://"):
    raise RuntimeError("workspace control endpoint must use HTTPS")
CONTROL_CERT = (os.environ["WORKSPACE_CONTROL_CERT"], os.environ["WORKSPACE_CONTROL_KEY"])
CA_BUNDLE = os.environ["INTERNAL_CA_BUNDLE"]
LOCKS: dict[str, tuple[int, int, str, float]] = {}

REQUEST_KEYS = {"schema_version", "request_id", "workspace_id", "operation", "payload"}
OPERATIONS = {
    "append_evidence", "observe", "lock", "activate", "reobserve",
    "invoke", "readback", "freeze", "unlock",
}

@dataclass(frozen=True)
class Peer:
    pid: int
    uid: int
    gid: int

def canonical(value: Any) -> bytes:
    return jcs.canonicalize(value)

def sha256(value: bytes) -> str:
    return "sha256:" + hashlib.sha256(value).hexdigest()

def strict_object(value: Any, keys: set[str], label: str) -> dict:
    if not isinstance(value, dict) or set(value) != keys:
        raise ValueError(label + " schema differs")
    return value

def recv_exact(conn: socket.socket, length: int) -> bytes:
    chunks: list[bytes] = []
    remaining = length
    while remaining:
        chunk = conn.recv(remaining)
        if not chunk:
            raise EOFError("truncated RPC frame")
        chunks.append(chunk)
        remaining -= len(chunk)
    return b"".join(chunks)

def recv_frame(conn: socket.socket) -> dict:
    length = struct.unpack("!I", recv_exact(conn, 4))[0]
    if length < 2 or length > MAX_FRAME:
        raise ValueError("RPC frame length outside policy")
    raw = recv_exact(conn, length)
    request = strict_object(json.loads(raw), REQUEST_KEYS, "request")
    if request["schema_version"] != "aidefend.desktop-observer-rpc.v1":
        raise ValueError("unsupported RPC schema")
    if ID.fullmatch(str(request["request_id"])) is None or ID.fullmatch(str(request["workspace_id"])) is None:
        raise ValueError("invalid request/workspace id")
    if request["operation"] not in OPERATIONS or not isinstance(request["payload"], dict):
        raise ValueError("invalid operation or payload")
    return request

def send_frame(conn: socket.socket, value: dict) -> None:
    raw = canonical(value)
    if len(raw) > MAX_FRAME:
        raise ValueError("response exceeds frame limit")
    conn.sendall(struct.pack("!I", len(raw)) + raw)

def peer_identity(conn: socket.socket) -> Peer:
    raw = conn.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize("3i"))
    return Peer(*struct.unpack("3i", raw))

def verify_envelope(path: Path, key: VerifyKey, expected_schema: str) -> tuple[dict, str]:
    descriptor = os.open(path, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
    try:
        info = os.fstat(descriptor)
        if not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o022:
            raise PermissionError("signed envelope ownership/mode invalid")
        if info.st_size < 32 or info.st_size > MAX_FRAME:
            raise ValueError("signed envelope size invalid")
        raw = os.read(descriptor, info.st_size + 1)
        if len(raw) != info.st_size:
            raise OSError("signed envelope changed during read")
    finally:
        os.close(descriptor)
    envelope = strict_object(json.loads(raw), {"payload", "signature_b64"}, "envelope")
    payload = envelope["payload"]
    if not isinstance(payload, dict) or payload.get("schema_version") != expected_schema:
        raise ValueError("signed payload schema differs")
    key.verify(canonical(payload), base64.b64decode(envelope["signature_b64"], validate=True))
    return payload, sha256(canonical(payload))

def peer_cgroup(pid: int) -> str:
    lines = Path(f"/proc/{pid}/cgroup").read_text(encoding="utf-8").splitlines()
    unified = [line.split(":", 2)[2] for line in lines if line.startswith("0::")]
    if len(unified) != 1:
        raise PermissionError("peer has no unique cgroup v2 identity")
    return unified[0]

def load_context(peer: Peer, workspace_id: str) -> tuple[dict, dict, str]:
    registry, registry_sha = verify_envelope(
        REGISTRY_DIR / (workspace_id + ".json"), REGISTRY_VERIFY_KEY,
        "aidefend.desktop-workspace-registry.v1",
    )
    registry_fields = {
        "schema_version", "workspace_id", "session_id", "task_id", "tenant_id",
        "agent_uid", "agent_cgroup", "workspace_cgroup",
        "dispatcher_uid", "dispatcher_cgroup", "policy_path",
        "policy_sha256", "launch_outcome", "launch_receipt_sha256",
        "state", "expires_unix",
    }
    if (
        set(registry) != registry_fields
        or registry.get("workspace_id") != workspace_id
        or registry.get("dispatcher_uid") != peer.uid
        or registry.get("dispatcher_cgroup") != peer_cgroup(peer.pid)
        or registry.get("state") != "RUNNING"
        or registry.get("launch_outcome") != "PASS"
        or not SHA256.fullmatch(str(registry.get("launch_receipt_sha256")))
        or time.time() > float(registry.get("expires_unix", 0))
    ):
        raise PermissionError("peer is not bound to active workspace registry")
    policy_path = Path(str(registry["policy_path"]))
    if policy_path.parent != Path("/etc/aidefend/desktop-policies"):
        raise PermissionError("policy path outside trusted directory")
    policy, policy_sha = verify_envelope(
        policy_path, POLICY_VERIFY_KEY, "aidefend.desktop-action-policy.v1"
    )
    if (
        set(policy) != {
            "schema_version", "policy_version", "valid_from_unix",
            "valid_until_unix", "targets", "effect_readback_url",
            "control_timeout_seconds", "accessibility_snapshot_node_limit",
            "control_response_max_bytes", "control_response_chunk_bytes",
        }
        or policy_sha != registry.get("policy_sha256")
        or not float(policy["valid_from_unix"]) <= time.time() <= float(policy["valid_until_unix"])
        or isinstance(policy["control_timeout_seconds"], bool)
        or not isinstance(policy["control_timeout_seconds"], (int, float))
        or not math.isfinite(float(policy["control_timeout_seconds"]))
        or float(policy["control_timeout_seconds"]) <= 0
        or isinstance(policy["accessibility_snapshot_node_limit"], bool)
        or not isinstance(policy["accessibility_snapshot_node_limit"], int)
        or policy["accessibility_snapshot_node_limit"] < 1
        or isinstance(policy["control_response_max_bytes"], bool)
        or not isinstance(policy["control_response_max_bytes"], int)
        or policy["control_response_max_bytes"] < 2
        or isinstance(policy["control_response_chunk_bytes"], bool)
        or not isinstance(policy["control_response_chunk_bytes"], int)
        or policy["control_response_chunk_bytes"] < 1
        or policy["control_response_chunk_bytes"] > policy["control_response_max_bytes"]
        or not str(policy["effect_readback_url"]).startswith("https://")
    ):
        raise PermissionError("registry/policy digest mismatch")
    return registry, policy, registry_sha

def database() -> sqlite3.Connection:
    EVIDENCE_DB.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    db = sqlite3.connect(EVIDENCE_DB, isolation_level=None)
    db.execute("PRAGMA journal_mode=WAL")
    db.execute("PRAGMA synchronous=FULL")
    db.execute(
        "CREATE TABLE IF NOT EXISTS receipts("
        "receipt_id TEXT PRIMARY KEY, receipt_bytes BLOB NOT NULL, signature BLOB NOT NULL)"
    )
    return db

def persist_signed(event: dict) -> dict:
    event = {
        **event, "service_identity": SERVICE_IDENTITY,
        "receipt_id": str(uuid4()),
        "recorded_at": datetime.now(timezone.utc).isoformat(),
    }
    event_bytes = canonical(event)
    signature = EVIDENCE_KEY.sign(event_bytes).signature
    with database() as db:
        db.execute("BEGIN IMMEDIATE")
        db.execute("INSERT INTO receipts VALUES(?,?,?)", (event["receipt_id"], event_bytes, signature))
        row = db.execute(
            "SELECT receipt_bytes,signature FROM receipts WHERE receipt_id=?", (event["receipt_id"],)
        ).fetchone()
        db.commit()
    if row is None or row[0] != event_bytes or row[1] != signature:
        raise OSError("durable receipt readback mismatch")
    EVIDENCE_KEY.verify_key.verify(row[0], row[1])
    return {
        "schema_version": "aidefend.desktop-observer-receipt.v1",
        "event": event,
        "receipt_sha256": sha256(row[0]),
        "signature_b64": base64.b64encode(row[1]).decode("ascii"),
    }

def app_in_workspace(app: Atspi.Accessible, registry: dict) -> bool:
    pid = int(app.get_process_id())
    return peer_cgroup(pid) == registry["workspace_cgroup"]

def walk(node: Atspi.Accessible, path: Sequence[int] = ()):
    yield node, path
    for index in range(node.get_child_count()):
        child = node.get_child_at_index(index)
        if child is not None:
            yield from walk(child, path + (index,))

def owning_window(node: Atspi.Accessible) -> tuple[Atspi.Accessible, int]:
    current = node
    parent_hops = 0
    while current is not None:
        if current.get_role_name() in {"frame", "window", "dialog"}:
            return current, parent_hops
        current = current.get_parent()
        parent_hops += 1
    raise LookupError("target has no owning accessibility window")

def accessibility_snapshot(node: Atspi.Accessible, limit: int) -> dict:
    rows: list[dict] = []
    for current, path in walk(node):
        if len(rows) >= limit:
            raise LookupError("accessibility subtree exceeds signed measurement limit")
        states = sorted(str(value) for value in current.get_state_set().get_states())
        rows.append({
            "path": list(path), "role": current.get_role_name(),
            "name": current.get_name(), "states": states,
        })
    return {"nodes": rows, "node_count": len(rows)}

def resolve_target(registry: dict, policy: dict, target_id: str) -> tuple[Atspi.Accessible, dict]:
    targets = policy.get("targets")
    if not isinstance(targets, dict) or target_id not in targets:
        raise LookupError("target id absent from signed policy")
    selector = strict_object(
        targets[target_id],
        {
            "application_executable_sha256", "window_name", "role_name", "accessible_name",
            "allowed_actions", "target_resource", "derived_effect",
            "invariant_id", "invariant_sha256", "expected_effect_sha256",
            "readback_source", "readback_adapter_id", "readback_adapter_version",
            "complete_population_required",
        },
        "target selector",
    )
    desktop = Atspi.get_desktop(0)
    matches: list[tuple[Atspi.Accessible, Sequence[int], int, Path, Atspi.Accessible, int]] = []
    for app_index in range(desktop.get_child_count()):
        app = desktop.get_child_at_index(app_index)
        if app is None or not app_in_workspace(app, registry):
            continue
        pid = int(app.get_process_id())
        exe = Path(f"/proc/{pid}/exe").resolve(strict=True)
        if sha256(exe.read_bytes()) != selector["application_executable_sha256"]:
            continue
        for node, path in walk(app):
            try:
                window, parent_hops = owning_window(node)
            except LookupError:
                continue
            if (
                node.get_role_name() == selector["role_name"]
                and node.get_name() == selector["accessible_name"]
                and window.get_name() == selector["window_name"]
            ):
                matches.append((node, path, pid, exe, window, parent_hops))
    if len(matches) != 1:
        raise LookupError("signed selector did not resolve exactly one element")
    node, path, pid, exe, window, parent_hops = matches[0]
    component = node.get_component_iface()
    rect = component.get_extents(Atspi.CoordType.SCREEN)
    process_start_ticks = Path(f"/proc/{pid}/stat").read_text().split()[21]
    subtree = accessibility_snapshot(node, policy["accessibility_snapshot_node_limit"])
    subtree_sha = sha256(canonical(subtree))
    window_depth = len(path) - parent_hops
    window_path = path[:window_depth]
    binding = {
        "tenant_id": registry["tenant_id"],
        "workspace_id": registry["workspace_id"],
        "session_id": registry["session_id"],
        "task_id": registry["task_id"],
        "foreground_process_identity": sha256(canonical({
            "pid": pid, "process_start_ticks": process_start_ticks,
            "executable_sha256": selector["application_executable_sha256"],
        })),
        "target_application": str(exe),
        "window_handle": "atspi:" + str(pid) + ":" + ".".join(str(index) for index in window_path),
        "window_title": window.get_name(),
        "accessibility_element_id": target_id,
        "accessibility_role": node.get_role_name(),
        "accessibility_name": node.get_name(),
        "accessibility_bounds": (rect.x, rect.y, rect.width, rect.height),
        "accessibility_subtree_sha256": subtree_sha,
        "pre_action_state_sha256": subtree_sha,
        "target_resource": selector["target_resource"],
        "derived_effect": selector["derived_effect"],
        "execution_policy_version": policy["policy_version"],
        "execution_policy_sha256": registry["policy_sha256"],
        "invariant_id": selector["invariant_id"],
        "invariant_sha256": selector["invariant_sha256"],
        "expected_effect_sha256": selector["expected_effect_sha256"],
        "readback_source": selector["readback_source"],
        "readback_adapter_id": selector["readback_adapter_id"],
        "readback_adapter_version": selector["readback_adapter_version"],
        "complete_population_required": selector["complete_population_required"],
    }
    return node, binding

def verify_consumed_h018(receipt: dict, binding: dict, action_digest: str) -> None:
    envelope = strict_object(receipt, {"payload", "signature_b64"}, "H-018 receipt")
    payload = envelope["payload"]
    H018_VERIFY_KEY.verify(canonical(payload), base64.b64decode(envelope["signature_b64"], validate=True))
    if (
        payload.get("schema_version") != "aidefend.h018-consumption-receipt.v1"
        or payload.get("outcome") != "PASS"
        or payload.get("state") != "CONSUMED"
        or payload.get("workspace_id") != binding["workspace_id"]
        or payload.get("task_id") != binding["task_id"]
        or payload.get("action_digest") != action_digest
        or payload.get("execution_binding_sha256") != sha256(canonical(binding))
        or time.time() > float(payload.get("expires_unix", 0))
    ):
        raise PermissionError("H-018 consumption receipt binding/state invalid")

def verify_remote_envelope(value: Any, key: VerifyKey, schema: str) -> tuple[dict, str]:
    envelope = strict_object(value, {"payload", "signature_b64"}, "remote receipt")
    payload = envelope["payload"]
    if not isinstance(payload, dict) or payload.get("schema_version") != schema:
        raise ValueError("remote receipt schema differs")
    body = canonical(payload)
    key.verify(body, base64.b64decode(envelope["signature_b64"], validate=True))
    return payload, sha256(body)

def bounded_response_json(response: requests.Response, maximum: int, chunk_size: int) -> Any:
    declared = response.headers.get("content-length")
    if declared is not None:
        try:
            declared_size = int(declared)
        except ValueError as exc:
            raise ValueError("invalid response content-length") from exc
        if declared_size < 0 or declared_size > maximum:
            raise ValueError("authority response exceeds signed policy")
    raw = bytearray()
    for chunk in response.iter_content(chunk_size=chunk_size):
        raw.extend(chunk)
        if len(raw) > maximum:
            raise ValueError("authority response exceeds signed policy")
    if not raw:
        raise ValueError("authority response is empty")
    return json.loads(bytes(raw))

def lock_workspace(peer: Peer, workspace_id: str) -> str:
    lock_path = REGISTRY_DIR / (workspace_id + ".lock")
    descriptor = os.open(lock_path, os.O_CREAT | os.O_RDWR | os.O_CLOEXEC, 0o600)
    fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
    lock_id = str(uuid4())
    LOCKS[lock_id] = (descriptor, peer.pid, workspace_id, time.monotonic() + 30.0)
    return lock_id

def require_lock(peer: Peer, workspace_id: str, lock_id: str) -> int:
    record = LOCKS.get(lock_id)
    if record is None or record[1:3] != (peer.pid, workspace_id) or time.monotonic() > record[3]:
        raise PermissionError("missing, foreign, or expired workspace lock")
    return record[0]

def unlock(peer: Peer, workspace_id: str, lock_id: str) -> None:
    descriptor = require_lock(peer, workspace_id, lock_id)
    fcntl.flock(descriptor, fcntl.LOCK_UN)
    os.close(descriptor)
    del LOCKS[lock_id]

def freeze(
    workspace_id: str,
    task_id: str,
    reason: str,
    timeout_seconds: float,
    response_max_bytes: int,
    response_chunk_bytes: int,
) -> dict:
    with requests.post(
        f"{CONTROL_URL}/v1/workspaces/{workspace_id}/freeze",
        json={"task_id": task_id, "reason": reason, "revoke_leases": True},
        cert=CONTROL_CERT, verify=CA_BUNDLE, timeout=timeout_seconds,
        allow_redirects=False, stream=True,
    ) as response:
        response.raise_for_status()
        value = bounded_response_json(
            response, response_max_bytes, response_chunk_bytes,
        )
    receipt, receipt_sha = verify_remote_envelope(
        value, CONTROL_VERIFY_KEY, "aidefend.workspace-freeze-receipt.v1"
    )
    if (
        receipt.get("workspace_id") != workspace_id
        or receipt.get("task_id") != task_id
        or receipt.get("state") != "FROZEN"
        or receipt.get("lease_state") != "REVOKED"
    ):
        raise RuntimeError("freeze receipt context/state invalid")
    return {**receipt, "receipt_sha256": receipt_sha}

def dispatch(peer: Peer, request: dict) -> dict:
    workspace_id = request["workspace_id"]
    operation = request["operation"]
    payload = request["payload"]
    registry, policy, registry_sha = load_context(peer, workspace_id)
    if operation in {"observe", "reobserve", "activate", "invoke", "readback"}:
        require_watch_readback(registry)
    if operation == "append_evidence":
        event = strict_object(payload, {"event"}, "append evidence")["event"]
        if not isinstance(event, dict) or event.get("control") != "AID-I-008.004":
            raise ValueError("evidence event control/schema invalid")
        return persist_signed({**event, "workspace_registry_sha256": registry_sha})
    if operation == "lock":
        return persist_signed({
            "control": "AID-I-008.004", "stage": "workspace_lock", "outcome": "PASS",
            "workspace_id": workspace_id, "lock_id": lock_workspace(peer, workspace_id),
        })
    if operation == "observe":
        target_id = str(payload.get("target_id", ""))
        _, binding = resolve_target(registry, policy, target_id)
        return persist_signed({
            "control": "AID-I-008.004", "stage": "observe", "outcome": "PASS",
            "binding": binding, "workspace_registry_sha256": registry_sha,
        })
    lock_id = str(payload.get("lock_id", ""))
    require_lock(peer, workspace_id, lock_id)
    if operation == "unlock":
        unlock(peer, workspace_id, lock_id)
        return persist_signed({
            "control": "AID-I-008.004", "stage": "unlock", "outcome": "PASS",
            "workspace_id": workspace_id, "lock_id": lock_id,
        })
    if operation == "freeze":
        freeze_receipt = freeze(
            workspace_id,
            registry["task_id"],
            str(payload.get("reason", "unspecified")),
            float(policy["control_timeout_seconds"]),
            int(policy["control_response_max_bytes"]),
            int(policy["control_response_chunk_bytes"]),
        )
        return persist_signed({
            "control": "AID-I-008.004", "stage": "freeze", "outcome": "PASS",
            "freeze_receipt": freeze_receipt,
        })
    target_id = str(payload.get("target_id", ""))
    node, binding = resolve_target(registry, policy, target_id)
    if operation == "reobserve":
        return persist_signed({
            "control": "AID-I-008.004", "stage": operation, "outcome": "PASS",
            "binding": binding, "workspace_registry_sha256": registry_sha,
        })
    if operation == "activate":
        if not node.get_component_iface().grab_focus():
            raise RuntimeError("AT-SPI focus activation failed")
        return persist_signed({
            "control": "AID-I-008.004", "stage": "activate", "outcome": "PASS",
            "binding_sha256": sha256(canonical(binding)),
        })
    if operation == "invoke":
        action_name = str(payload.get("action_name", ""))
        target = policy["targets"][target_id]
        if action_name not in target["allowed_actions"]:
            raise PermissionError("action absent from signed allowlist")
        action_digest = str(payload.get("action_digest", ""))
        verify_consumed_h018(payload.get("h018_consumption_receipt"), binding, action_digest)
        action_iface = node.get_action_iface()
        indices = [
            index for index in range(action_iface.get_n_actions())
            if action_iface.get_action_name(index) == action_name
        ]
        if len(indices) != 1 or not action_iface.do_action(indices[0]):
            raise RuntimeError("stable AT-SPI invocation failed")
        return persist_signed({
            "control": "AID-I-008.004", "stage": "invoke", "outcome": "PASS",
            "action_digest": action_digest, "binding_sha256": sha256(canonical(binding)),
            "action_name": action_name,
        })
    if operation == "readback":
        effect_url = str(policy["effect_readback_url"]).rstrip("/")
        with requests.post(
            effect_url,
            json={
                "workspace_id": workspace_id, "task_id": registry["task_id"],
                "target_resource": binding["target_resource"],
                "invariant_sha256": binding["invariant_sha256"],
                "action_id": payload.get("action_id"),
                "action_digest": payload.get("action_digest"),
                "invocation_receipt_sha256": payload.get("invocation_receipt_sha256"),
            },
            cert=CONTROL_CERT,
            verify=CA_BUNDLE,
            timeout=float(policy["control_timeout_seconds"]),
            allow_redirects=False,
            stream=True,
        ) as response:
            response.raise_for_status()
            value = bounded_response_json(
                response, policy["control_response_max_bytes"],
                policy["control_response_chunk_bytes"],
            )
        readback, readback_sha = verify_remote_envelope(
            value, READBACK_VERIFY_KEY,
            "aidefend.desktop-effect-readback.v1",
        )
        if (
            readback.get("workspace_id") != workspace_id
            or readback.get("task_id") != registry["task_id"]
            or readback.get("action_id") != payload.get("action_id")
            or readback.get("action_digest") != payload.get("action_digest")
            or readback.get("target_resource") != binding["target_resource"]
            or readback.get("invariant_sha256") != binding["invariant_sha256"]
            or readback.get("expected_effect_sha256") != binding["expected_effect_sha256"]
            or readback.get("readback_source") != binding["readback_source"]
            or readback.get("readback_adapter_id") != binding["readback_adapter_id"]
            or readback.get("readback_adapter_version") != binding["readback_adapter_version"]
            or readback.get("outcome") not in {"PASS", "FAIL", "INSUFFICIENT_DATA", "ERROR"}
            or readback.get("population_complete") is not True
            or not SHA256.fullmatch(str(readback.get("measured_population_sha256")))
            or not isinstance(readback.get("measured_population_count"), int)
            or readback.get("measured_population_count") < 1
            or (
                readback.get("outcome") == "PASS"
                and readback.get("observed_effect_sha256") != binding["expected_effect_sha256"]
            )
        ):
            raise RuntimeError("readback receipt schema/context incomplete")
        readback = {**readback, "signed_readback_receipt_sha256": readback_sha}
        return persist_signed({
            "control": "AID-I-008.004", "stage": "effect_readback",
            "outcome": readback.get("outcome"), "binding_sha256": sha256(canonical(binding)),
            "readback": readback,
        })
    raise ValueError("unsupported operation")

def serve_connection(conn: socket.socket) -> None:
    peer = peer_identity(conn)
    request_id = "unknown"
    workspace_id = "unknown"
    try:
        request = recv_frame(conn)
        request_id = request["request_id"]
        workspace_id = request["workspace_id"]
        receipt = dispatch(peer, request)
    except PermissionError as exc:
        receipt = persist_signed({
            "control": "AID-I-008.004", "stage": "observer_rpc", "outcome": "FAIL",
            "request_id": request_id, "workspace_id": workspace_id,
            "peer": asdict(peer), "reason": type(exc).__name__,
        })
    except (FileNotFoundError, LookupError) as exc:
        receipt = persist_signed({
            "control": "AID-I-008.004", "stage": "observer_rpc",
            "outcome": "INSUFFICIENT_DATA", "request_id": request_id,
            "workspace_id": workspace_id, "peer": asdict(peer),
            "reason": type(exc).__name__,
        })
    except Exception as exc:
        receipt = persist_signed({
            "control": "AID-I-008.004", "stage": "observer_rpc", "outcome": "ERROR",
            "request_id": request_id, "workspace_id": workspace_id,
            "peer": asdict(peer), "reason": type(exc).__name__,
        })
    send_frame(conn, {"schema_version": "aidefend.desktop-observer-rpc-response.v1", "receipt": receipt})

def main() -> None:
    Path(SOCKET_PATH).parent.mkdir(parents=True, exist_ok=True, mode=0o750)
    try:
        os.unlink(SOCKET_PATH)
    except FileNotFoundError:
        pass
    listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    listener.bind(SOCKET_PATH)
    os.chmod(SOCKET_PATH, 0o660)
    listener.listen(64)
    while True:
        conn, _ = listener.accept()
        with conn:
            serve_connection(conn)

if __name__ == "__main__":
    main()

Client and service verification

Use the same four-byte network-order frame in desktop_clients.py. The client must verify the service UID from SO_PEERCRED, the receipt's Ed25519 signature, exact schema, request/workspace/action/binding fields, outcome enum, and receipt digest; it then independently reads the receipt by ID through a read-only evidence endpoint and compares bytes. The client never supplies application, process, window, policy, or observed-state facts. It supplies only a workspace ID, signed-policy target ID, intended action, and—at invocation—the exact consumed H-018 receipt. Run a pre-deployment integration test that launches a disposable workspace, resolves a known AT-SPI control, locks it, activates and re-resolves it, attempts one invocation with an invalid receipt (must be denied), performs one approved invocation, validates authoritative readback, freezes, unlocks, and verifies every durable receipt.

Require fresh watch-mode readback before every observation or action

Install the verify-only desktop_watch_readiness.py client in the observer and screenshot services. It holds only the independently pinned watchdog policy and evidence public keys plus the protected runtime profile. It does not import the privileged watchdog module, open its spool, or possess its signing key or mTLS credentials. For each observation/action it requests a fresh nonce-bound sample from the root-owned G005 Unix socket, authenticates the server with Linux SO_PEERCRED, and verifies signature, task/workspace, current policy, runtime, collector generation, completeness and freshness. The G005 worker performs the protected append, fsync and exact readback before returning its signed acknowledgement; the client independently verifies that acknowledgement, while storage durability remains the worker's attested property. Missing or invalid acknowledgement keeps the caller blocked. The model workspace has neither socket access nor any service credentials.

# File: computer_use/desktop_watch_readiness.py
from __future__ import annotations
import base64
import hashlib
import json
import math
import os
import re
import secrets
import socket
import stat
import struct
import time
from datetime import datetime
from pathlib import Path
import jcs
from nacl.signing import VerifyKey

# Only public trust and a protected deployment profile enter this process.
# Do not import watch_mode: it belongs exclusively to the privileged worker.
POLICY_PATH = Path("/etc/aidefend/watchdog-policy.signed.json")
SOCKET_PATH = Path("/run/computer-use-watchdog/readback.sock")
POLICY_VERIFY_KEY = VerifyKey(bytes.fromhex(os.environ["WATCHDOG_POLICY_VERIFY_KEY_HEX"]))
WATCH_READBACK_KEY = VerifyKey(bytes.fromhex(os.environ["WATCHDOG_READBACK_VERIFY_KEY_HEX"]))
RUNTIME_PROFILE_VERSION = os.environ["AIDEFEND_RUNTIME_PROFILE_VERSION"]
RUNTIME_PROFILE_SHA256 = os.environ["AIDEFEND_RUNTIME_PROFILE_SHA256"]
POLICY_MAX_BYTES = int(os.environ["WATCHDOG_POLICY_MAX_BYTES"])
OPAQUE_ID = re.compile(r"[A-Za-z0-9._:-]{1,128}")
SHA256 = re.compile(r"sha256:[0-9a-f]{64}")
if (POLICY_MAX_BYTES < 1 or not RUNTIME_PROFILE_VERSION
        or re.fullmatch(r"[a-f0-9]{64}", RUNTIME_PROFILE_SHA256) is None):
    raise ValueError("invalid observer watch bootstrap profile")


def canonical(value):
    try:
        return jcs.canonicalize(value)
    except (RecursionError, OverflowError) as exc:
        raise ValueError("watchdog JSON exceeds supported depth or numeric range") from exc


def strict_json(raw):
    def unique(pairs):
        result = {}
        for key, value in pairs:
            if key in result: raise ValueError("duplicate JSON key")
            result[key] = value
        return result
    def nonfinite(_value): raise ValueError("nonfinite JSON value")
    try:
        return json.loads(raw, object_pairs_hook=unique, parse_constant=nonfinite)
    except (RecursionError, OverflowError) as exc:
        raise ValueError("watchdog JSON exceeds supported depth or numeric range") from exc


def current_policy():
    fd = os.open(POLICY_PATH, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
    try:
        info = os.fstat(fd)
        if (not stat.S_ISREG(info.st_mode) or info.st_uid != 0
                or info.st_mode & 0o022 or not 1 <= info.st_size <= POLICY_MAX_BYTES):
            raise PermissionError("observer watchdog policy ownership/size differs")
        raw = os.read(fd, info.st_size + 1)
        if len(raw) != info.st_size: raise OSError("watch policy changed during read")
    finally: os.close(fd)
    value = strict_json(raw)
    if not isinstance(value, dict) or set(value) != {"payload", "signature_b64"}:
        raise ValueError("signed watchdog policy envelope differs")
    policy = value["payload"]
    if not isinstance(policy, dict): raise ValueError("watchdog policy is missing")
    body = canonical(policy)
    POLICY_VERIFY_KEY.verify(body, base64.b64decode(value["signature_b64"], validate=True))
    if policy.get("schema_version") != "aidefend.desktop-watchdog-policy.v2":
        raise ValueError("watchdog policy requires the readback-client v2 contract")
    for name in ("valid_from_unix", "valid_until_unix", "readback_timeout_seconds",
                 "telemetry_max_age_seconds", "telemetry_max_future_skew_seconds"):
        number = policy[name]
        if type(number) not in (int, float) or not math.isfinite(number):
            raise ValueError("watchdog policy numeric bound differs")
    if not policy["valid_from_unix"] <= time.time() <= policy["valid_until_unix"]:
        raise PermissionError("watchdog policy is not currently valid")
    if (policy["readback_timeout_seconds"] <= 0 or policy["telemetry_max_age_seconds"] <= 0
            or policy["telemetry_max_future_skew_seconds"] < 0
            or type(policy["http_response_max_bytes"]) is not int
            or policy["http_response_max_bytes"] < 2):
        raise ValueError("watchdog transport/freshness bounds differ")
    uids = policy["readback_client_uids"]
    if (not isinstance(uids, list) or not uids
            or any(type(uid) is not int or uid < 1 for uid in uids) or len(set(uids)) != len(uids)):
        raise ValueError("watchdog readback service identities differ")
    return policy, "sha256:" + hashlib.sha256(body).hexdigest()


def peer_uid(stream):
    # Linux SO_PEERCRED authenticates the kernel peer, not a request field.
    if not hasattr(socket, "SO_PEERCRED"):
        raise OSError("this transport requires Linux SO_PEERCRED")
    return struct.unpack("3i", stream.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))[1]


def read_exact(stream, count, deadline):
    result = bytearray()
    while len(result) < count:
        remaining = deadline - time.monotonic()
        if remaining <= 0: raise TimeoutError("watchdog RPC deadline exceeded")
        stream.settimeout(remaining)
        chunk = stream.recv(count - len(result))
        if not chunk: raise EOFError("watchdog RPC frame is truncated")
        result.extend(chunk)
    return bytes(result)


def read_frame(stream, maximum, deadline):
    count = struct.unpack("!I", read_exact(stream, 4, deadline))[0]
    if not 1 <= count <= maximum: raise ValueError("watchdog RPC frame exceeds policy")
    return strict_json(read_exact(stream, count, deadline))


def write_frame(stream, value, maximum, deadline):
    body = canonical(value)
    if not 1 <= len(body) <= maximum: raise ValueError("watchdog RPC frame exceeds policy")
    remaining = deadline - time.monotonic()
    if remaining <= 0: raise TimeoutError("watchdog RPC deadline exceeded")
    stream.settimeout(remaining)
    stream.sendall(struct.pack("!I", len(body)) + body)


def connect_worker(deadline):
    parent, info = SOCKET_PATH.parent.lstat(), SOCKET_PATH.lstat()
    if (not stat.S_ISDIR(parent.st_mode) or parent.st_uid != 0 or parent.st_mode & 0o022
            or not stat.S_ISSOCK(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o007):
        raise PermissionError("watchdog socket custody differs")
    stream = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    try:
        remaining = deadline - time.monotonic()
        if remaining <= 0: raise TimeoutError("watchdog RPC deadline exceeded")
        stream.settimeout(remaining)
        stream.connect(str(SOCKET_PATH))
        if peer_uid(stream) != 0: raise PermissionError("watchdog RPC peer is not the privileged worker")
        return stream
    except Exception:
        stream.close()
        raise


def signed_record(envelope):
    if not isinstance(envelope, dict) or set(envelope) != {"record", "record_sha256", "signature_b64"}:
        raise ValueError("watch readback envelope schema differs")
    record = envelope["record"]
    if not isinstance(record, dict): raise ValueError("watch readback record is missing")
    body = canonical(record)
    WATCH_READBACK_KEY.verify(body, base64.b64decode(envelope["signature_b64"], validate=True))
    if envelope["record_sha256"] != "sha256:" + hashlib.sha256(body).hexdigest():
        raise ValueError("watch readback digest differs")
    return record


def verify_readback(envelope, context, nonce, policy, policy_sha):
    record = signed_record(envelope)
    fields = {"schema_version", "control", "stage", "outcome", "record_id", "policy_version", "policy_sha256",
              "runtime_profile_version", "runtime_profile_sha256", "spooled_at", "process_monotonic_ns", "payload"}
    if (set(record) != fields or record["schema_version"] != "aidefend.computer_use_watch_spool.v1"
            or record["control"] != "AID-I-008.004" or record["stage"] != "watch_sample" or record["outcome"] != "PASS"
            or record["policy_version"] != policy["policy_version"] or record["policy_sha256"] != policy_sha
            or record["runtime_profile_version"] != RUNTIME_PROFILE_VERSION
            or record["runtime_profile_sha256"] != RUNTIME_PROFILE_SHA256):
        raise PermissionError("watch sample is not a current successful policy-bound acknowledgement")
    timestamp = datetime.fromisoformat(record["spooled_at"].replace("Z", "+00:00"))
    if timestamp.tzinfo is None: raise ValueError("watch timestamp needs a timezone")
    age = time.time() - timestamp.timestamp()
    if (not math.isfinite(age) or age > policy["telemetry_max_age_seconds"]
            or age < -policy["telemetry_max_future_skew_seconds"]):
        raise FileNotFoundError("watch acknowledgement is stale or future dated")
    payload = record["payload"]
    payload_fields = {"task_id", "workspace_id", "telemetry_receipt_sha256", "collector_identity", "collector_generation",
                      "measured_population_sha256", "measured_population_count", "elapsed_seconds", "policy_denials",
                      "challenge_nonce"}
    if (not isinstance(payload, dict) or set(payload) != payload_fields
            or payload["task_id"] != context["task_id"] or payload["workspace_id"] != context["workspace_id"]
            or payload["challenge_nonce"] != nonce or payload["collector_identity"] != policy["collector_identity"]
            or payload["collector_generation"] != policy["collector_generation"]
            or type(payload["measured_population_count"]) is not int or payload["measured_population_count"] < 1
            or SHA256.fullmatch(str(payload["measured_population_sha256"])) is None
            or SHA256.fullmatch(str(payload["telemetry_receipt_sha256"])) is None
            or type(payload["elapsed_seconds"]) not in (int, float) or not math.isfinite(payload["elapsed_seconds"])
            or not 0 <= payload["elapsed_seconds"] <= policy["max_task_seconds"]
            or type(payload["policy_denials"]) is not int or not 0 <= payload["policy_denials"] < policy["max_policy_denials"]):
        raise PermissionError("watch sample challenge, workspace, collector or population differs")
    return envelope["record_sha256"]


def require_watch_readback(context):
    for field in ("task_id", "workspace_id"):
        if not isinstance(context[field], str) or OPAQUE_ID.fullmatch(context[field]) is None:
            raise ValueError("invalid watch readiness context")
    policy, policy_sha = current_policy()
    nonce = secrets.token_hex(32)
    request = {"schema_version": "aidefend.desktop-watch-readback-request.v1", "task_id": context["task_id"],
               "workspace_id": context["workspace_id"], "challenge_nonce": nonce}
    deadline = time.monotonic() + policy["readback_timeout_seconds"]
    with connect_worker(deadline) as stream:
        write_frame(stream, request, policy["http_response_max_bytes"], deadline)
        envelope = read_frame(stream, policy["http_response_max_bytes"], deadline)
    current, current_sha = current_policy()
    if current_sha != policy_sha or current != policy:
        raise PermissionError("watchdog policy rotated during readiness sampling")
    return verify_readback(envelope, context, nonce, policy, policy_sha)

A successful synchronous sample establishes readiness at that boundary only. Deploy an independent continuously supervised G005 worker to cover idle/model-computation periods. Its liveness, full telemetry coverage and terminal containment evidence are necessary for the full desktop-control claim; these snippets do not establish that the background supervisor is deployed.

Checker deployment

Run observer/screenshot services as authorized non-root host service UIDs. For watchdog readiness they need read access to the root-owned signed policy and OS-authorized connection access to the G005 socket, but no watchdog spool descriptor, watchdog signing key or watchdog control/audit mTLS private key. The privileged G005 worker alone owns those watchdog files; observer/screenshot evidence signing and control access remain their separately scoped existing responsibilities. Keep the protected socket and all service identities outside the model workspace. Missing permissions or acknowledgement cannot release an observation. The regression fixture verifies real producer signatures, challenge bindings and framed client/worker exchange with explicit transport/UID adapters; it does not establish deployed Linux identities or continuous supervision.

AID-I-008.004-G002

Concept:

A desktop computer-use agent should not attach to a user's daily desktop session. Start every task in a disposable workspace: no inherited browser profile, no password manager, no long-lived enterprise app session, no unscoped home directory, and no global clipboard reuse. The workspace launcher should receive a task policy and enforce it before the model sees the first screenshot.

Step 1: Declare the desktop workspace policy

# File: policy/computer_use_workspace.yaml
workspace:
  mode: disposable
  logged_out_mode: true
  max_task_minutes: 30
  allowed_applications:
    - chromium
    - libreoffice
    - file-viewer
  denied_applications:
    - password-manager
    - terminal
    - cloud-admin-console
  allowed_directories:
    - /workspace/task-input
    - /workspace/task-output
  denied_directories:
    - /home/user
    - /mnt/corporate-shares
  clipboard:
    mode: task_scoped_broker
    allow_global_clipboard: false
  network:
    egress_policy: allowlisted
    allowed_domains:
      - docs.internal.corp
      - tickets.internal.corp
  screenshots:
    redact_secrets: true
    injection_scan_required: true

Step 2: Enforce the policy before session start

# File: computer_use/workspace_lint.py
from __future__ import annotations

import hashlib as _aidefend_policy_hashlib
import json as _aidefend_policy_json
import math as _aidefend_policy_math
import os as _aidefend_policy_os
import subprocess as _aidefend_policy_subprocess
import tempfile as _aidefend_policy_tempfile
from pathlib import Path as _AidefendPolicyPath


def _aidefend_reject_duplicate_keys(pairs):
    value = {}
    for key, item in pairs:
        if key in value:
            raise ValueError(f"duplicate signed-policy JSON key: {key}")
        value[key] = item
    return value


def _aidefend_reject_nonfinite(value):
    raise ValueError(f"non-finite signed-policy JSON value: {value}")


def _aidefend_load_signed_policy(*, path_env, signature_env, key_env, schema_version):
    policy_path = _AidefendPolicyPath(_aidefend_policy_os.environ[path_env])
    signature_path = _AidefendPolicyPath(_aidefend_policy_os.environ[signature_env])
    verify_timeout = float(_aidefend_policy_os.environ["AIDEFEND_POLICY_VERIFY_TIMEOUT_SECONDS"])
    if not _aidefend_policy_math.isfinite(verify_timeout) or verify_timeout <= 0:
        raise RuntimeError("policy-verification timeout is invalid")
    payload = policy_path.read_bytes()
    signature = signature_path.read_bytes()
    if not payload or not signature:
        raise RuntimeError("signed policy payload or signature is empty")
    with _aidefend_policy_tempfile.TemporaryDirectory(prefix="aidefend-policy-") as directory:
        root = _AidefendPolicyPath(directory)
        _aidefend_policy_os.chmod(root, 0o700)
        payload_path = root / "policy.json"
        signature_path = root / "policy.sig"
        payload_path.write_bytes(payload)
        signature_path.write_bytes(signature)
        _aidefend_policy_os.chmod(payload_path, 0o400)
        _aidefend_policy_os.chmod(signature_path, 0o400)
        _aidefend_policy_subprocess.run(
            ["cosign", "verify-blob", "--key", _aidefend_policy_os.environ[key_env],
             "--bundle", str(signature_path), str(payload_path)],
            check=True, capture_output=True, text=True, timeout=verify_timeout,
        )
        verified = payload_path.read_bytes()
        if verified != payload:
            raise RuntimeError("verified policy snapshot changed")
    policy = _aidefend_policy_json.loads(
        verified.decode("utf-8", errors="strict"),
        object_pairs_hook=_aidefend_reject_duplicate_keys,
        parse_constant=_aidefend_reject_nonfinite,
    )
    if (not isinstance(policy, dict) or policy.get("schema_version") != schema_version
            or not isinstance(policy.get("policy_version"), str) or not policy["policy_version"]):
        raise RuntimeError("signature-verified policy schema or version is invalid")
    if "_aidefend_verified_sha256" in policy:
        raise RuntimeError("signed policy uses a reserved evidence field")
    policy["_aidefend_verified_sha256"] = _aidefend_policy_hashlib.sha256(verified).hexdigest()
    return policy


import ipaddress
import json
import re
import sys
from pathlib import Path, PurePosixPath

import yaml


WORKSPACE_LINT_POLICY = _aidefend_load_signed_policy(
    path_env="I008004_WORKSPACE_LINT_POLICY_PATH",
    signature_env="I008004_WORKSPACE_LINT_POLICY_SIGNATURE",
    key_env="I008004_WORKSPACE_LINT_POLICY_VERIFY_KEY",
    schema_version="aidefend.desktop-workspace-lint-policy.v1",
)
MAX_TASK_MINUTES = WORKSPACE_LINT_POLICY.get("maximum_task_minutes")
if isinstance(MAX_TASK_MINUTES, bool) or not isinstance(MAX_TASK_MINUTES, int) or MAX_TASK_MINUTES < 1:
    raise RuntimeError("signed workspace maximum task duration is invalid")
REQUIRED_DENIED_APPS = {"password-manager", "terminal", "cloud-admin-console"}
APPROVED_DIRECTORY_ROOTS = (
    PurePosixPath("/workspace/task-input"),
    PurePosixPath("/workspace/task-output"),
)
WORKSPACE_FIELDS = {
    "mode", "logged_out_mode", "max_task_minutes", "allowed_applications",
    "denied_applications", "allowed_directories", "denied_directories",
    "clipboard", "network", "screenshots",
}
APP_NAME = re.compile(r"[a-z0-9][a-z0-9._-]{0,63}")
DNS_NAME = re.compile(
    r"(?=.{1,253}$)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,63}",
    re.IGNORECASE,
)


class NoDuplicateSafeLoader(yaml.SafeLoader):
    pass


def construct_unique_mapping(loader, node, deep=False):
    mapping = {}
    for key_node, value_node in node.value:
        key = loader.construct_object(key_node, deep=deep)
        if key in mapping:
            raise yaml.constructor.ConstructorError(
                "while constructing workspace policy", node.start_mark,
                f"duplicate key: {key}", key_node.start_mark,
            )
        mapping[key] = loader.construct_object(value_node, deep=deep)
    return mapping


NoDuplicateSafeLoader.add_constructor(
    yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
    construct_unique_mapping,
)


def require_mapping(value, field: str, errors: list[str]) -> dict:
    if not isinstance(value, dict):
        errors.append(f"{field} must be a mapping")
        return {}
    return value


def require_string_list(value, field: str, errors: list[str]) -> list[str]:
    if not isinstance(value, list) or not value:
        errors.append(f"{field} must be a nonempty list")
        return []
    if any(not isinstance(item, str) or not item.strip() for item in value):
        errors.append(f"{field} must contain only nonempty strings")
        return []
    normalized = [item.strip() for item in value]
    if len(normalized) != len(set(normalized)):
        errors.append(f"{field} must not contain duplicates")
    return normalized


def is_within(path: PurePosixPath, root: PurePosixPath) -> bool:
    return path == root or root in path.parents


def validate_directories(values: list[str], field: str, errors: list[str]) -> list[PurePosixPath]:
    parsed = []
    for value in values:
        path = PurePosixPath(value)
        if not path.is_absolute() or ".." in path.parts or "~" in path.parts or "\x00" in value:
            errors.append(f"{field} contains unsafe path: {value}")
            continue
        parsed.append(path)
        if field == "allowed_directories" and not any(is_within(path, root) for root in APPROVED_DIRECTORY_ROOTS):
            errors.append(f"allowed directory is outside approved task roots: {value}")
    return parsed


def valid_domain(value: str) -> bool:
    if "*" in value or "://" in value or "/" in value or ":" in value:
        return False
    try:
        ipaddress.ip_address(value)
        return False
    except ValueError:
        return DNS_NAME.fullmatch(value) is not None


def lint_workspace_policy(path: str) -> dict:
    errors: list[str] = []
    try:
        loaded = yaml.load(Path(path).read_text(encoding="utf-8"), Loader=NoDuplicateSafeLoader)
    except (OSError, yaml.YAMLError) as exc:
        return {"passed": False, "errors": [f"policy cannot be safely loaded: {exc}"]}

    root = require_mapping(loaded, "document", errors)
    if set(root) != {"workspace"}:
        errors.append("document must contain only the workspace mapping")
    policy = require_mapping(root.get("workspace"), "workspace", errors)
    unknown_fields = sorted(set(policy) - WORKSPACE_FIELDS)
    missing_fields = sorted(WORKSPACE_FIELDS - set(policy))
    if unknown_fields:
        errors.append("unknown workspace fields: " + ", ".join(unknown_fields))
    if missing_fields:
        errors.append("missing workspace fields: " + ", ".join(missing_fields))

    if policy.get("mode") != "disposable":
        errors.append("workspace must be disposable")
    if policy.get("logged_out_mode") is not True:
        errors.append("logged_out_mode must be true")
    duration = policy.get("max_task_minutes")
    if isinstance(duration, bool) or not isinstance(duration, int) or not 1 <= duration <= MAX_TASK_MINUTES:
        errors.append(f"max_task_minutes must be an integer from 1 to {MAX_TASK_MINUTES}")

    allowed_apps = require_string_list(policy.get("allowed_applications"), "allowed_applications", errors)
    denied_apps = require_string_list(policy.get("denied_applications"), "denied_applications", errors)
    invalid_apps = sorted({item for item in allowed_apps + denied_apps if APP_NAME.fullmatch(item) is None})
    if invalid_apps:
        errors.append("invalid application identifiers: " + ", ".join(invalid_apps))
    overlap_apps = sorted(set(allowed_apps) & set(denied_apps))
    if overlap_apps:
        errors.append("applications cannot be both allowed and denied: " + ", ".join(overlap_apps))
    missing_denied = sorted(REQUIRED_DENIED_APPS - set(denied_apps))
    if missing_denied:
        errors.append("missing denied applications: " + ", ".join(missing_denied))

    allowed_dir_values = require_string_list(policy.get("allowed_directories"), "allowed_directories", errors)
    denied_dir_values = require_string_list(policy.get("denied_directories"), "denied_directories", errors)
    allowed_dirs = validate_directories(allowed_dir_values, "allowed_directories", errors)
    denied_dirs = validate_directories(denied_dir_values, "denied_directories", errors)
    for allowed in allowed_dirs:
        for denied in denied_dirs:
            if is_within(allowed, denied) or is_within(denied, allowed):
                errors.append(f"allowed and denied directory scopes overlap: {allowed} / {denied}")

    clipboard = require_mapping(policy.get("clipboard"), "clipboard", errors)
    if set(clipboard) != {"mode", "allow_global_clipboard"}:
        errors.append("clipboard must declare only mode and allow_global_clipboard")
    if clipboard.get("mode") != "task_scoped_broker":
        errors.append("clipboard mode must be task_scoped_broker")
    if clipboard.get("allow_global_clipboard") is not False:
        errors.append("global clipboard must be disabled")

    network = require_mapping(policy.get("network"), "network", errors)
    if set(network) != {"egress_policy", "allowed_domains"}:
        errors.append("network must declare only egress_policy and allowed_domains")
    if network.get("egress_policy") != "allowlisted":
        errors.append("network egress_policy must be allowlisted")
    domains = require_string_list(network.get("allowed_domains"), "network.allowed_domains", errors)
    invalid_domains = sorted(value for value in domains if not valid_domain(value))
    if invalid_domains:
        errors.append("domains must be exact DNS names without wildcard, URL, port, or IP: " + ", ".join(invalid_domains))

    screenshots = require_mapping(policy.get("screenshots"), "screenshots", errors)
    if set(screenshots) != {"redact_secrets", "injection_scan_required"}:
        errors.append("screenshots must declare only redact_secrets and injection_scan_required")
    if screenshots.get("redact_secrets") is not True:
        errors.append("screenshot secret redaction must be required")
    if screenshots.get("injection_scan_required") is not True:
        errors.append("screenshot injection scan must be required")

    return {
        "passed": not errors,
        "errors": errors,
        "lint_policy_version": WORKSPACE_LINT_POLICY["policy_version"],
        "lint_policy_sha256": WORKSPACE_LINT_POLICY["_aidefend_verified_sha256"],
    }


def main() -> None:
    result = lint_workspace_policy(sys.argv[1])
    print(json.dumps(result, indent=2, sort_keys=True))
    raise SystemExit(0 if result["passed"] else 2)


if __name__ == "__main__":
    main()

Step 3: Launch, measure, sign, and independently read back the disposable workspace

The YAML linter is only pre-signing admission. A privileged release job signs the exact YAML digest together with task/workspace identity, a digest-pinned OCI desktop image, a pre-attested egress profile, host input/output roots, agent UID, expiration, and launcher version. The following launcher consumes that signed envelope, starts a rootless Podman container with a nested headless Wayland compositor, measures the running instance, writes the observer's signed workspace registry, and emits a non-overwrite signed launch receipt. The image must contain only the approved GUI applications and the computer-use supervisor; it must not contain a host shell, password manager, inherited browser profile, or host display socket.

# File: computer_use/disposable_workspace_launcher.py
from __future__ import annotations

import base64
import hashlib
import json
import math
import os
import re
import sqlite3
import stat
import subprocess
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from uuid import uuid4

import jcs
import yaml
from nacl.exceptions import BadSignatureError
from nacl.signing import SigningKey, VerifyKey

from computer_use.workspace_lint import lint_workspace_policy

POLICY = Path("/etc/aidefend/launch-requests/computer-use.signed.json")
WORKSPACE_YAML = Path("/etc/aidefend/launch-requests/computer_use_workspace.yaml")
REGISTRY_DIR = Path("/run/aidefend/desktop-workspaces")
EVIDENCE_DB = Path("/var/lib/aidefend/workspace-launcher/evidence.sqlite3")
POLICY_KEY = VerifyKey(bytes.fromhex(os.environ["WORKSPACE_LAUNCH_POLICY_VERIFY_KEY_HEX"]))
NETWORK_KEY = VerifyKey(bytes.fromhex(os.environ["EGRESS_PROFILE_VERIFY_KEY_HEX"]))
SUPERVISOR_KEY = VerifyKey(bytes.fromhex(os.environ["DESKTOP_SUPERVISOR_VERIFY_KEY_HEX"]))
REGISTRY_KEY = SigningKey(bytes.fromhex(os.environ["WORKSPACE_REGISTRY_SIGNING_KEY_HEX"]))
EVIDENCE_KEY = SigningKey(bytes.fromhex(os.environ["WORKSPACE_EVIDENCE_SIGNING_KEY_HEX"]))
IMAGE = re.compile(r"[a-z0-9./_-]+@sha256:[0-9a-f]{64}")
IDENTIFIER = re.compile(r"[A-Za-z0-9._:-]{1,128}")
SHA256 = re.compile(r"sha256:[0-9a-f]{64}")
OUTCOMES = {"PASS", "FAIL", "INSUFFICIENT_DATA", "NOT_APPLICABLE", "ERROR"}

def canonical(value: Any) -> bytes:
    return jcs.canonicalize(value)

def digest(raw: bytes) -> str:
    return "sha256:" + hashlib.sha256(raw).hexdigest()

def read_regular(path: Path, maximum: int) -> bytes:
    descriptor = os.open(path, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
    try:
        info = os.fstat(descriptor)
        if not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o022:
            raise PermissionError("launch input ownership/mode invalid")
        if info.st_size < 1 or info.st_size > maximum:
            raise ValueError("launch input size invalid")
        raw = os.read(descriptor, info.st_size + 1)
        if len(raw) != info.st_size:
            raise OSError("launch input changed during read")
        return raw
    finally:
        os.close(descriptor)

def verify_envelope(raw: bytes, key: VerifyKey, schema: str) -> dict:
    envelope = json.loads(raw)
    if not isinstance(envelope, dict) or set(envelope) != {"payload", "signature_b64"}:
        raise ValueError("signed envelope schema differs")
    payload = envelope["payload"]
    if not isinstance(payload, dict) or payload.get("schema_version") != schema:
        raise ValueError("signed payload schema differs")
    key.verify(canonical(payload), base64.b64decode(envelope["signature_b64"], validate=True))
    return payload

def evidence_db() -> sqlite3.Connection:
    EVIDENCE_DB.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    db = sqlite3.connect(EVIDENCE_DB, isolation_level=None)
    db.execute("PRAGMA journal_mode=WAL")
    db.execute("PRAGMA synchronous=FULL")
    db.execute(
        "CREATE TABLE IF NOT EXISTS launch_receipts("
        "receipt_id TEXT PRIMARY KEY, body BLOB NOT NULL, signature BLOB NOT NULL)"
    )
    return db

def emit(outcome: str, stage: str, **facts: Any) -> dict:
    if outcome not in OUTCOMES:
        raise ValueError("invalid outcome")
    event = {
        "schema_version": "aidefend.desktop-workspace-launch-receipt.v1",
        "control": "AID-I-008.004", "receipt_id": str(uuid4()),
        "outcome": outcome, "stage": stage,
        "recorded_at": datetime.now(timezone.utc).isoformat(), **facts,
    }
    body = canonical(event)
    signature = EVIDENCE_KEY.sign(body).signature
    with evidence_db() as db:
        db.execute("BEGIN IMMEDIATE")
        db.execute("INSERT INTO launch_receipts VALUES(?,?,?)", (event["receipt_id"], body, signature))
        row = db.execute(
            "SELECT body,signature FROM launch_receipts WHERE receipt_id=?", (event["receipt_id"],)
        ).fetchone()
        db.commit()
    if row is None or row[0] != body or row[1] != signature:
        raise OSError("launch receipt durable readback differs")
    EVIDENCE_KEY.verify_key.verify(row[0], row[1])
    return {
        "event": event, "receipt_sha256": digest(body),
        "signature_b64": base64.b64encode(signature).decode("ascii"),
    }

def run(*args: str, timeout_seconds: float) -> str:
    result = subprocess.run(
        list(args), check=True, text=True, encoding="utf-8",
        stdout=subprocess.PIPE, stderr=subprocess.PIPE,
        timeout=timeout_seconds,
        env={"PATH": "/usr/sbin:/usr/bin:/sbin:/bin", "LANG": "C.UTF-8"},
    )
    return result.stdout.strip()

def namespace_identity(pid: int, name: str) -> str:
    return os.readlink(f"/proc/{pid}/ns/{name}")

def write_registry(payload: dict) -> tuple[Path, str]:
    signature = REGISTRY_KEY.sign(canonical(payload)).signature
    envelope = {
        "payload": payload,
        "signature_b64": base64.b64encode(signature).decode("ascii"),
    }
    raw = canonical(envelope)
    REGISTRY_DIR.mkdir(parents=True, exist_ok=True, mode=0o750)
    path = REGISTRY_DIR / (payload["workspace_id"] + ".json")
    descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC, 0o600)
    try:
        if os.write(descriptor, raw) != len(raw):
            raise OSError("short workspace registry write")
        os.fsync(descriptor)
    finally:
        os.close(descriptor)
    dir_fd = os.open(REGISTRY_DIR, os.O_RDONLY | os.O_DIRECTORY)
    try:
        os.fsync(dir_fd)
    finally:
        os.close(dir_fd)
    if read_regular(path, 1_048_576) != raw:
        raise OSError("workspace registry readback differs")
    return path, digest(canonical(payload))

def launch(execution_surface: str) -> dict:
    if execution_surface != "desktop_gui":
        return emit(
            "NOT_APPLICABLE", "applicability", execution_surface=execution_surface,
            reason="no desktop GUI workspace requested",
        )
    cid = None
    try:
        if not POLICY.exists() or not WORKSPACE_YAML.exists():
            return emit(
                "INSUFFICIENT_DATA", "admission",
                reason="signed launch request or exact workspace YAML missing",
            )
        yaml_raw = read_regular(WORKSPACE_YAML, 262_144)
        lint = lint_workspace_policy(str(WORKSPACE_YAML))
        if not lint["passed"]:
            return emit("FAIL", "policy_lint", policy_errors=lint["errors"])
        workspace_config = yaml.safe_load(yaml_raw)["workspace"]
        request = verify_envelope(
            read_regular(POLICY, 1_048_576), POLICY_KEY,
            "aidefend.desktop-workspace-launch-request.v1",
        )
        required = {
            "schema_version", "workspace_id", "session_id", "task_id", "tenant_id",
            "workspace_policy_sha256", "desktop_action_policy_sha256",
            "desktop_image", "network_profile",
            "network_profile_receipt", "input_host_path", "output_host_path",
            "agent_uid", "dispatcher_uid", "dispatcher_cgroup",
            "launcher_version", "expires_unix", "workspace_lint_policy_sha256",
            "command_timeout_seconds", "cleanup_timeout_seconds",
        }
        if set(request) != required:
            return emit("FAIL", "admission", reason="launch request fields differ")
        if (
            IDENTIFIER.fullmatch(str(request["workspace_id"])) is None
            or IDENTIFIER.fullmatch(str(request["session_id"])) is None
            or IDENTIFIER.fullmatch(str(request["task_id"])) is None
            or request["workspace_policy_sha256"] != digest(yaml_raw)
            or request["workspace_lint_policy_sha256"] != lint["lint_policy_sha256"]
            or SHA256.fullmatch(str(request["desktop_action_policy_sha256"])) is None
            or IMAGE.fullmatch(str(request["desktop_image"])) is None
            or time.time() > float(request["expires_unix"])
            or not isinstance(request["agent_uid"], int)
            or request["agent_uid"] < 10000
            or not isinstance(request["dispatcher_uid"], int)
            or request["dispatcher_uid"] < 1
            or not isinstance(request["dispatcher_cgroup"], str)
            or not request["dispatcher_cgroup"].startswith("/")
            or any(
                isinstance(request[field], bool)
                or not isinstance(request[field], (int, float))
                or not math.isfinite(float(request[field]))
                or float(request[field]) <= 0
                for field in (
                    "command_timeout_seconds",
                    "cleanup_timeout_seconds",
                )
            )
        ):
            return emit("FAIL", "admission", reason="signed launch bindings invalid")
        command_timeout_seconds = float(request["command_timeout_seconds"])
        cleanup_timeout_seconds = float(request["cleanup_timeout_seconds"])
        launch_request_sha256 = digest(canonical(request))
        network = verify_envelope(
            canonical(request["network_profile_receipt"]), NETWORK_KEY,
            "aidefend.desktop-egress-profile-receipt.v1",
        )
        if (
            network.get("profile_name") != request["network_profile"]
            or network.get("outcome") != "PASS"
            or network.get("policy_sha256") != request["workspace_policy_sha256"]
            or time.time() > float(network.get("expires_unix", 0))
        ):
            return emit("FAIL", "network_admission", reason="egress profile receipt mismatch")
        workspace_id = request["workspace_id"]
        input_path = Path(request["input_host_path"]).resolve(strict=True)
        output_path = Path(request["output_host_path"]).resolve(strict=True)
        if (
            Path("/srv/aidefend/task-inputs") not in input_path.parents
            or Path("/srv/aidefend/task-outputs") not in output_path.parents
        ):
            return emit("FAIL", "admission", reason="host mount outside task roots")
        cidfile = f"/run/aidefend/{workspace_id}.cid"
        run(
            "podman", "run", "--detach", "--rm",
            "--name", "computer-use-" + workspace_id,
            "--cidfile", cidfile,
            "--read-only", "--cap-drop=ALL", "--security-opt=no-new-privileges",
            "--user", str(request["agent_uid"]),
            "--ipc=private", "--pid=private",
            "--network", request["network_profile"],
            "--tmpfs", "/home/agent:rw,nosuid,nodev,noexec,size=256m",
            "--tmpfs", "/run/user:rw,nosuid,nodev,noexec,size=64m",
            "--mount", f"type=bind,src={input_path},dst=/workspace/task-input,ro=true,relabel=private",
            "--mount", f"type=bind,src={output_path},dst=/workspace/task-output,rw=true,relabel=private",
            "--env", "HOME=/home/agent", "--env", "XDG_RUNTIME_DIR=/run/user",
            "--env", "WAYLAND_DISPLAY=wayland-0",
            "--label", "aidefend.workspace_id=" + workspace_id,
            "--label", "aidefend.policy_sha256=" + request["workspace_policy_sha256"],
            request["desktop_image"],
            "/usr/local/bin/computer-use-supervisor", "--nested-wayland", "--logged-out",
            timeout_seconds=command_timeout_seconds,
        )
        cid = read_regular(Path(cidfile), 256).decode("ascii").strip()
        if re.fullmatch(r"[0-9a-f]{64}", cid) is None:
            raise RuntimeError("invalid Podman container id")
        inspect_rows = json.loads(run(
            "podman", "inspect", cid,
            timeout_seconds=command_timeout_seconds,
        ))
        if not isinstance(inspect_rows, list) or len(inspect_rows) != 1:
            raise RuntimeError("Podman inspect population differs")
        inspect = inspect_rows[0]
        supervisor = verify_envelope(
            run(
                "podman", "exec", "--user", "0", cid,
                "/usr/local/bin/computer-use-supervisor", "attest", "--format=json",
                timeout_seconds=command_timeout_seconds,
            ).encode("utf-8"),
            SUPERVISOR_KEY, "aidefend.desktop-supervisor-attestation.v1",
        )
        supervisor_sha = digest(canonical(supervisor))
        if set(supervisor) != {
            "schema_version", "workspace_id", "logged_out",
            "nested_wayland_isolated", "global_clipboard_access",
            "clipboard_broker_mode", "secret_redaction_enabled",
            "screenshot_gate_enabled", "running_applications",
            "process_population_complete", "process_population_count",
            "process_population_sha256",
        }:
            raise RuntimeError("desktop supervisor attestation fields differ")
        state = inspect.get("State", {})
        host = inspect.get("HostConfig", {})
        config = inspect.get("Config", {})
        mounts = inspect.get("Mounts", [])
        pid = int(state.get("Pid", 0))
        measurements = {
            "container_id": cid,
            "running": state.get("Running") is True,
            "image_digest": inspect.get("ImageDigest") or config.get("Image"),
            "read_only_rootfs": host.get("ReadonlyRootfs") is True,
            "cap_drop": sorted(host.get("CapDrop") or []),
            "security_opt": sorted(host.get("SecurityOpt") or []),
            "network_mode": host.get("NetworkMode"),
            "user": config.get("User"),
            "mounts": sorted(
                ({
                    "destination": item.get("Destination"),
                    "source": item.get("Source"),
                    "rw": item.get("RW"),
                }
                for item in mounts),
                key=lambda item: (str(item["destination"]), str(item["source"]), bool(item["rw"])),
            ),
            "pid_namespace": namespace_identity(pid, "pid"),
            "mount_namespace": namespace_identity(pid, "mnt"),
            "network_namespace": namespace_identity(pid, "net"),
            "host_pid_namespace": namespace_identity(os.getpid(), "pid"),
            "host_mount_namespace": namespace_identity(os.getpid(), "mnt"),
            "host_network_namespace": namespace_identity(os.getpid(), "net"),
        }
        expected_mounts = {
            ("/workspace/task-input", str(input_path), False),
            ("/workspace/task-output", str(output_path), True),
        }
        observed_mounts = {
            (item["destination"], item["source"], item["rw"]) for item in measurements["mounts"]
            if item["destination"] in {"/workspace/task-input", "/workspace/task-output"}
        }
        passed = (
            measurements["running"]
            and measurements["image_digest"] in {
                request["desktop_image"], request["desktop_image"].rsplit("@", 1)[1]
            }
            and measurements["read_only_rootfs"]
            and "ALL" in measurements["cap_drop"]
            and "no-new-privileges" in " ".join(measurements["security_opt"])
            and measurements["network_mode"] == request["network_profile"]
            and measurements["user"] == str(request["agent_uid"])
            and observed_mounts == expected_mounts
            and supervisor["workspace_id"] == workspace_id
            and supervisor["logged_out"] is True
            and supervisor["nested_wayland_isolated"] is True
            and supervisor["global_clipboard_access"] is False
            and supervisor["clipboard_broker_mode"] == "task_scoped_broker"
            and supervisor["secret_redaction_enabled"] is True
            and supervisor["screenshot_gate_enabled"] is True
            and isinstance(supervisor["running_applications"], list)
            and all(isinstance(item, str) and item for item in supervisor["running_applications"])
            and set(supervisor["running_applications"]).issubset(
                set(workspace_config["allowed_applications"])
            )
            and supervisor["process_population_complete"] is True
            and isinstance(supervisor["process_population_count"], int)
            and supervisor["process_population_count"] >= 1
            and SHA256.fullmatch(str(supervisor["process_population_sha256"])) is not None
            and measurements["pid_namespace"] != measurements["host_pid_namespace"]
            and measurements["mount_namespace"] != measurements["host_mount_namespace"]
            and measurements["network_namespace"] != measurements["host_network_namespace"]
        )
        population = [
            {"kind": "podman_inspect", "sha256": digest(canonical(inspect))},
            {"kind": "namespace_measurements", "sha256": digest(canonical(measurements))},
            {"kind": "egress_profile_receipt", "sha256": digest(canonical(network))},
            {"kind": "desktop_supervisor_attestation", "sha256": supervisor_sha},
        ]
        if not passed:
            run(
                "podman", "rm", "--force", cid,
                timeout_seconds=cleanup_timeout_seconds,
            )
            return emit(
                "FAIL", "runtime_measurement", workspace_id=workspace_id,
                measurements=measurements, measured_population=population,
                measured_population_count=len(population),
                measured_population_sha256=digest(canonical(population)),
            )
        cgroup = Path(f"/proc/{pid}/cgroup").read_text(encoding="utf-8")
        cgroup_paths = [line.split(":", 2)[2] for line in cgroup.splitlines() if line.startswith("0::")]
        if len(cgroup_paths) != 1:
            raise RuntimeError("workspace has no unique cgroup v2 identity")
        provisional = emit(
            "PASS", "runtime_measurement", workspace_id=workspace_id,
            task_id=request["task_id"], tenant_id=request["tenant_id"],
            launcher_version=request["launcher_version"],
            policy_sha256=request["workspace_policy_sha256"],
            workspace_lint_policy_sha256=request["workspace_lint_policy_sha256"],
            workspace_lint_policy_version=lint["lint_policy_version"],
            launch_request_sha256=launch_request_sha256,
            command_timeout_seconds=command_timeout_seconds,
            cleanup_timeout_seconds=cleanup_timeout_seconds,
            measurements=measurements, measured_population=population,
            measured_population_count=len(population),
            measured_population_sha256=digest(canonical(population)),
        )
        registry = {
            "schema_version": "aidefend.desktop-workspace-registry.v1",
            "workspace_id": workspace_id, "session_id": request["session_id"],
            "task_id": request["task_id"],
            "tenant_id": request["tenant_id"], "agent_uid": request["agent_uid"],
            "agent_cgroup": cgroup_paths[0], "workspace_cgroup": cgroup_paths[0],
            "dispatcher_uid": request["dispatcher_uid"],
            "dispatcher_cgroup": request["dispatcher_cgroup"],
            "policy_path": f"/etc/aidefend/desktop-policies/{workspace_id}.json",
            "policy_sha256": request["desktop_action_policy_sha256"],
            "launch_outcome": "PASS",
            "launch_receipt_sha256": provisional["receipt_sha256"],
            "state": "RUNNING", "expires_unix": request["expires_unix"],
        }
        registry_path, registry_sha = write_registry(registry)
        return emit(
            "PASS", "workspace_prepared", workspace_id=workspace_id,
            launch_receipt_sha256=provisional["receipt_sha256"],
            registry_path=str(registry_path), registry_sha256=registry_sha,
            measured_population_sha256=digest(canonical(population)),
            measured_population_count=len(population),
            launch_request_sha256=launch_request_sha256,
            command_timeout_seconds=command_timeout_seconds,
            cleanup_timeout_seconds=cleanup_timeout_seconds,
        )
    except FileNotFoundError as exc:
        return emit("INSUFFICIENT_DATA", "launcher", reason=type(exc).__name__)
    except (PermissionError, ValueError, BadSignatureError) as exc:
        if cid:
            subprocess.run(
                ["podman", "rm", "--force", cid],
                check=False,
                timeout=cleanup_timeout_seconds,
            )
        return emit("FAIL", "launcher", reason=type(exc).__name__)
    except Exception as exc:
        if cid:
            subprocess.run(
                ["podman", "rm", "--force", cid],
                check=False,
                timeout=cleanup_timeout_seconds,
            )
        return emit("ERROR", "launcher", reason=type(exc).__name__)

if __name__ == "__main__":
    print(json.dumps(launch(os.environ.get("EXECUTION_SURFACE", "")), sort_keys=True))

Make applicability and numeric truth manifest-owned

Remove the caller argument and EXECUTION_SURFACE environment lookup from launch(). Verify the root-owned signed launch/deployment manifest first, require its closed-schema execution_surface field, and derive NOT_APPLICABLE only when that verified field is not desktop_gui. A missing or unverifiable manifest is INSUFFICIENT_DATA, never N/A. Reject Python Boolean values anywhere an integer or float is expected, and reject NaN/infinity before every time comparison.

def finite_number(value, field, *, minimum=None):
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError(field + " must be a signed JSON number")
    result = float(value)
    if not math.isfinite(result) or (minimum is not None and result < minimum):
        raise ValueError(field + " is outside the signed numeric domain")
    return result

# After signature verification and exact-field validation:
execution_surface = request["execution_surface"]
if not isinstance(execution_surface, str):
    return emit("FAIL", "applicability", reason="execution surface type invalid")
if execution_surface != "desktop_gui":
    return emit(
        "NOT_APPLICABLE", "applicability",
        launch_request_sha256=digest(canonical(request)),
        execution_surface=execution_surface,
        reason="signed deployment manifest has no desktop GUI path",
    )
expires_unix = finite_number(request["expires_unix"], "expires_unix", minimum=0)
network_expires_unix = finite_number(
    network["expires_unix"], "network.expires_unix", minimum=0
)
if time.time() > expires_unix or time.time() > network_expires_unix:
    return emit("FAIL", "admission", reason="launch or network receipt expired")
if type(request["agent_uid"]) is not int or type(request["dispatcher_uid"]) is not int:
    return emit("FAIL", "admission", reason="workspace UID type invalid")

Add execution_surface to the signed request's exact required-field set and remove the earlier pre-verification applicability branch.

Reconcile workspace removal and evidence uncertainty

Reserve a durable workspace-operation row before podman run and advance it through LAUNCHING, RUNNING_UNRELEASED, RUNNING, REMOVAL_PENDING, and DESTROYED. Before every cleanup attempt write REMOVAL_PENDING; after podman rm --force, independently require podman inspect to report no such container and confirm the workspace lease is revoked. A command timeout, nonzero removal, surviving container, or unreadable registry leaves the row retryable and blocked. If final evidence or registry publication fails after launch, retain the container ID and signed launch bindings in that row, revoke admission, and run the same reconciler; never leave a signed RUNNING record or discard the only live handle. The environment's short workspace lease is a backstop, not proof of removal.

Prepared workspace is not observation authorization

The launch receipt and RUNNING registry prove the measured disposable-workspace facet. They do not release a screenshot, authorize an action or establish a complete desktop-control PASS. The trusted dispatcher must use G001/G003, which verify a fresh signed G005 watch sample before exposing desktop observations, and G004 for action binding. Keep all model-visible screen and OS interfaces behind those services, and require independent continuous supervision for idle periods.

AID-I-008.004-G003

Required platform service and inputs

Apply this gate before a high-risk desktop action when the acting model has consumed one or more newly rendered frames. Require a trusted capture broker to sign the complete frame manifest and a separately credentialed classifier or human-review service to issue its decision. Missing dependencies, unmeasured frames, classifier abstention without completed human review, or incomplete population are INSUFFICIENT_DATA and block the action. Byte-bind and evaluate every frame under one signed gate policy with the pinned classifier version and, when policy requires, an independent human; insert the resulting signed decision once and read it back independently. Re-verify upstream capture and classifier receipts as inputs, then require all local bindings and coverage checks before the gate succeeds.

Signed gate policy and frame manifest

{
  "schema_version": "aidefend.desktop-screenshot-gate-policy.v1",
  "policy_version": "desktop-screen-gate/2026.07",
  "valid_from_unix": 1784160000,
  "valid_until_unix": 1786752000,
  "max_frame_age_seconds": 5,
  "max_future_skew_seconds": 2,
  "high_risk_actions": ["send", "delete", "approve", "change_access", "purchase"],
  "max_frames": 12,
  "max_frame_bytes": 12582912,
  "prefilter_patterns": [
    "ignore (all )?(previous|prior) instructions",
    "click (approve|allow|yes|confirm)",
    "send (the )?(file|secret|token|password)"
  ],
  "classifier_endpoint": "https://screen-classifier.internal.corp/v1/classify",
  "classifier_id": "desktop-injection-classifier",
  "classifier_version": "4.2.0",
  "classifier_model_sha256": "sha256:9f4b6f3dc41e7fd6a8f20cbaec698fb41ed778799548d5e1b6de4e1cf45adf2e",
  "allow_max_score": 0.15,
  "block_min_score": 0.65,
  "human_review_endpoint": "https://screen-review.internal.corp/v1/reviews",
  "human_review_required_between_thresholds": true,
  "service_timeout_seconds": 5,
  "human_review_timeout_seconds": 120,
  "service_response_max_bytes": 1048576,
  "service_response_chunk_bytes": 65536
}

The capture broker manifest must bind tenant, workspace, task, action candidate, monotonically ordered frame IDs, exact screenshot byte digests/sizes, capture timestamps, display IDs, and the complete-population digest/count. It signs the manifest outside the model workspace. Thresholds, endpoints, versions, population limits, and validity come from the signed policy; they are not universal constants in application code.

Production gate

# File: computer_use/signed_screenshot_gate.py
from __future__ import annotations

import base64
import hashlib
import io
import json
import math
import os
import re
import sqlite3
import stat
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from uuid import uuid4

import jcs
import pytesseract
import requests
from PIL import Image
from nacl.exceptions import BadSignatureError
from nacl.signing import SigningKey, VerifyKey
from computer_use.desktop_watch_readiness import require_watch_readback

POLICY_KEY = VerifyKey(bytes.fromhex(os.environ["SCREEN_GATE_POLICY_VERIFY_KEY_HEX"]))
CAPTURE_KEY = VerifyKey(bytes.fromhex(os.environ["SCREEN_CAPTURE_VERIFY_KEY_HEX"]))
CLASSIFIER_KEY = VerifyKey(bytes.fromhex(os.environ["SCREEN_CLASSIFIER_VERIFY_KEY_HEX"]))
HUMAN_KEY = VerifyKey(bytes.fromhex(os.environ["SCREEN_HUMAN_REVIEW_VERIFY_KEY_HEX"]))
EVIDENCE_KEY = SigningKey(bytes.fromhex(os.environ["SCREEN_GATE_EVIDENCE_SIGNING_KEY_HEX"]))
CLIENT_CERT = (os.environ["SCREEN_GATE_CLIENT_CERT"], os.environ["SCREEN_GATE_CLIENT_KEY"])
CA_BUNDLE = os.environ["INTERNAL_CA_BUNDLE"]
EVIDENCE_DB = Path("/var/lib/aidefend/screenshot-gate/evidence.sqlite3")
SHA256 = re.compile(r"sha256:[0-9a-f]{64}")
ID = re.compile(r"[A-Za-z0-9._:-]{1,128}")
OUTCOMES = {"PASS", "FAIL", "INSUFFICIENT_DATA", "NOT_APPLICABLE", "ERROR"}

def canonical(value: Any) -> bytes:
    return jcs.canonicalize(value)

def sha256(raw: bytes) -> str:
    return "sha256:" + hashlib.sha256(raw).hexdigest()

def verify_envelope(value: Any, key: VerifyKey, schema: str) -> tuple[dict, str]:
    if not isinstance(value, dict) or set(value) != {"payload", "signature_b64"}:
        raise ValueError("signed envelope schema differs")
    payload = value["payload"]
    if not isinstance(payload, dict) or payload.get("schema_version") != schema:
        raise ValueError("signed payload schema differs")
    body = canonical(payload)
    key.verify(body, base64.b64decode(value["signature_b64"], validate=True))
    return payload, sha256(body)

def read_json(path: Path, maximum: int) -> dict:
    descriptor = os.open(path, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
    try:
        info = os.fstat(descriptor)
        if not stat.S_ISREG(info.st_mode) or info.st_mode & 0o022:
            raise PermissionError("JSON evidence ownership/mode invalid")
        if info.st_size < 2 or info.st_size > maximum:
            raise ValueError("JSON evidence size invalid")
        raw = os.read(descriptor, info.st_size + 1)
        if len(raw) != info.st_size:
            raise OSError("JSON evidence changed during read")
        return json.loads(raw)
    finally:
        os.close(descriptor)

def read_frame(path: Path, expected_size: int, expected_sha: str, maximum: int) -> bytes:
    descriptor = os.open(path, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
    try:
        info = os.fstat(descriptor)
        if (
            not stat.S_ISREG(info.st_mode)
            or info.st_mode & 0o022
            or info.st_size != expected_size
            or info.st_size > maximum
        ):
            raise PermissionError("frame type/mode/size invalid")
        raw = os.read(descriptor, info.st_size + 1)
        if len(raw) != info.st_size or sha256(raw) != expected_sha:
            raise ValueError("frame bytes differ from signed manifest")
        image = Image.open(io.BytesIO(raw))
        image.verify()
        if image.format not in {"PNG", "JPEG", "WEBP"}:
            raise ValueError("unsupported screenshot encoding")
        return raw
    finally:
        os.close(descriptor)

def capture_epoch(value: object, field: str) -> float:
    if not isinstance(value, str):
        raise ValueError(field + " must be an ISO-8601 string")
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        raise ValueError(field + " must include a timezone")
    epoch = parsed.timestamp()
    if not math.isfinite(epoch):
        raise ValueError(field + " timestamp is non-finite")
    return epoch

def db() -> sqlite3.Connection:
    EVIDENCE_DB.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    connection = sqlite3.connect(EVIDENCE_DB, isolation_level=None)
    connection.execute("PRAGMA journal_mode=WAL")
    connection.execute("PRAGMA synchronous=FULL")
    connection.execute(
        "CREATE TABLE IF NOT EXISTS decisions("
        "decision_id TEXT PRIMARY KEY, body BLOB NOT NULL, signature BLOB NOT NULL)"
    )
    return connection

def emit(outcome: str, stage: str, **facts: Any) -> dict:
    if outcome not in OUTCOMES:
        raise ValueError("invalid outcome")
    event = {
        "schema_version": "aidefend.desktop-screenshot-gate-decision.v1",
        "control": "AID-I-008.004", "decision_id": str(uuid4()),
        "outcome": outcome, "stage": stage,
        "decided_at": datetime.now(timezone.utc).isoformat(), **facts,
    }
    body = canonical(event)
    signature = EVIDENCE_KEY.sign(body).signature
    with db() as connection:
        connection.execute("BEGIN IMMEDIATE")
        connection.execute("INSERT INTO decisions VALUES(?,?,?)", (event["decision_id"], body, signature))
        row = connection.execute(
            "SELECT body,signature FROM decisions WHERE decision_id=?", (event["decision_id"],)
        ).fetchone()
        connection.commit()
    if row is None or row[0] != body or row[1] != signature:
        raise OSError("gate decision durable readback differs")
    EVIDENCE_KEY.verify_key.verify(row[0], row[1])
    return {
        "event": event, "decision_sha256": sha256(body),
        "signature_b64": base64.b64encode(signature).decode("ascii"),
    }

def post_signed(
    endpoint: str, payload: dict, timeout: float, maximum: int, chunk_size: int,
) -> dict:
    with requests.post(
        endpoint, json=payload, cert=CLIENT_CERT, verify=CA_BUNDLE, timeout=timeout,
        headers={"Idempotency-Key": payload["request_id"]},
        allow_redirects=False, stream=True,
    ) as response:
        response.raise_for_status()
        declared = response.headers.get("content-length")
        if declared is not None:
            try:
                declared_size = int(declared)
            except ValueError as exc:
                raise ValueError("invalid service content-length") from exc
            if declared_size < 0 or declared_size > maximum:
                raise ValueError("service response exceeds signed policy")
        raw = bytearray()
        for chunk in response.iter_content(chunk_size=chunk_size):
            raw.extend(chunk)
            if len(raw) > maximum:
                raise ValueError("service response exceeds signed policy")
    if not raw:
        raise ValueError("service returned an empty response")
    value = json.loads(bytes(raw))
    if not isinstance(value, dict):
        raise ValueError("service did not return signed envelope")
    return value

def gate(
    *,
    authenticated_actor_id: str,
    policy_path: Path,
    manifest_path: Path,
) -> dict:
    try:
        policy, policy_sha = verify_envelope(
            read_json(policy_path, 1_048_576), POLICY_KEY,
            "aidefend.desktop-screenshot-gate-policy.v1",
        )
        required_policy = {
            "schema_version", "policy_version", "valid_from_unix",
            "valid_until_unix", "max_frame_age_seconds",
            "max_future_skew_seconds", "high_risk_actions", "max_frames",
            "max_frame_bytes", "prefilter_patterns", "classifier_endpoint",
            "classifier_id", "classifier_version", "classifier_model_sha256",
            "allow_max_score", "block_min_score", "human_review_endpoint",
            "human_review_required_between_thresholds",
            "service_timeout_seconds", "human_review_timeout_seconds",
            "service_response_max_bytes", "service_response_chunk_bytes",
        }
        if set(policy) != required_policy:
            raise ValueError("signed screenshot-gate policy schema differs")
        numeric_policy_fields = {
            "valid_from_unix": 0.0, "valid_until_unix": 0.0,
            "max_frame_age_seconds": 0.0, "max_future_skew_seconds": 0.0,
        }
        for field, minimum in numeric_policy_fields.items():
            value = policy[field]
            if (
                isinstance(value, bool)
                or not isinstance(value, (int, float))
                or not math.isfinite(float(value))
                or float(value) < minimum
            ):
                raise ValueError("signed screenshot policy number invalid: " + field)
        if float(policy["max_frame_age_seconds"]) <= 0:
            raise ValueError("signed frame age must be positive")
        now = time.time()
        if not float(policy["valid_from_unix"]) <= now <= float(policy["valid_until_unix"]):
            return emit("INSUFFICIENT_DATA", "policy", policy_sha256=policy_sha, reason="policy not currently valid")
        for field in ("service_timeout_seconds", "human_review_timeout_seconds"):
            value = policy.get(field)
            if (
                isinstance(value, bool)
                or not isinstance(value, (int, float))
                or not math.isfinite(float(value))
                or float(value) <= 0
            ):
                raise ValueError("signed policy timeout invalid: " + field)
        for field in ("max_frames", "max_frame_bytes", "service_response_max_bytes", "service_response_chunk_bytes"):
            value = policy[field]
            if isinstance(value, bool) or not isinstance(value, int) or value < 1:
                raise ValueError("signed policy positive integer invalid: " + field)
        if (
            policy["service_response_max_bytes"] < 2
            or policy["service_response_chunk_bytes"] > policy["service_response_max_bytes"]
            or SHA256.fullmatch(str(policy["classifier_model_sha256"])) is None
            or not str(policy["classifier_endpoint"]).startswith("https://")
            or not str(policy["human_review_endpoint"]).startswith("https://")
            or not isinstance(policy["prefilter_patterns"], list)
            or not policy["prefilter_patterns"]
            or any(not isinstance(pattern, str) or not pattern for pattern in policy["prefilter_patterns"])
        ):
            raise ValueError("signed policy endpoint, digest, pattern, or response bound invalid")
        if any(
            isinstance(policy[field], bool)
            or not isinstance(policy[field], (int, float))
            for field in ("allow_max_score", "block_min_score")
        ):
            raise ValueError("signed classifier threshold type is invalid")
        allow_score = float(policy["allow_max_score"])
        block_score = float(policy["block_min_score"])
        if (
            not math.isfinite(allow_score) or not math.isfinite(block_score)
            or not 0.0 <= allow_score < block_score <= 1.0
        ):
            raise ValueError("signed classifier thresholds are invalid")
        if not authenticated_actor_id or ID.fullmatch(authenticated_actor_id) is None:
            return emit("FAIL", "identity", reason="authenticated actor id invalid")
        manifest, manifest_sha = verify_envelope(
            read_json(manifest_path, 1_048_576), CAPTURE_KEY,
            "aidefend.desktop-frame-manifest.v1",
        )
        required_manifest = {
            "schema_version", "tenant_id", "workspace_id", "task_id", "action_id",
            "authenticated_actor_id", "execution_surface", "action_type",
            "frames", "population_complete", "measured_population_count",
            "measured_population_sha256", "captured_at",
        }
        if set(manifest) != required_manifest or manifest["population_complete"] is not True:
            return emit("INSUFFICIENT_DATA", "frame_population", manifest_sha256=manifest_sha, reason="manifest incomplete")
        manifest_capture_epoch = capture_epoch(manifest["captured_at"], "manifest.captured_at")
        if (
            now - manifest_capture_epoch > float(policy["max_frame_age_seconds"])
            or manifest_capture_epoch - now > float(policy["max_future_skew_seconds"])
        ):
            return emit("INSUFFICIENT_DATA", "frame_population", manifest_sha256=manifest_sha, reason="frame manifest is stale or future-dated")
        if manifest["authenticated_actor_id"] != authenticated_actor_id:
            return emit("FAIL", "identity", manifest_sha256=manifest_sha, reason="actor differs from trusted capture manifest")
        execution_surface = manifest["execution_surface"]
        action_type = manifest["action_type"]
        if execution_surface != "desktop_gui":
            return emit(
                "NOT_APPLICABLE", "applicability", manifest_sha256=manifest_sha,
                execution_surface=execution_surface,
                reason="trusted capture manifest identifies a non-desktop surface",
            )
        require_watch_readback(manifest)
        if action_type not in policy["high_risk_actions"]:
            return emit(
                "NOT_APPLICABLE", "applicability", manifest_sha256=manifest_sha,
                action_type=action_type,
                reason="trusted capture manifest identifies an action outside the signed high-risk set",
            )
        frames = manifest["frames"]
        if (
            not isinstance(frames, list) or not frames
            or len(frames) > int(policy["max_frames"])
            or isinstance(manifest["measured_population_count"], bool)
            or not isinstance(manifest["measured_population_count"], int)
            or manifest["measured_population_count"] != len(frames)
        ):
            return emit("INSUFFICIENT_DATA", "frame_population", manifest_sha256=manifest_sha, reason="frame count invalid")
        expected_frame_fields = {
            "frame_id", "sequence", "path", "screenshot_sha256",
            "size_bytes", "display_id", "captured_at",
        }
        frame_ids = [item.get("frame_id") for item in frames if isinstance(item, dict)]
        if (
            len(frame_ids) != len(frames)
            or any(set(item) != expected_frame_fields for item in frames)
            or any(not isinstance(value, str) or ID.fullmatch(value) is None for value in frame_ids)
            or len(set(frame_ids)) != len(frame_ids)
            or any(isinstance(item["sequence"], bool) or not isinstance(item["sequence"], int) for item in frames)
            or any(isinstance(item["size_bytes"], bool) or not isinstance(item["size_bytes"], int) or item["size_bytes"] < 1 or item["size_bytes"] > policy["max_frame_bytes"] for item in frames)
            or any(not isinstance(item["path"], str) or not item["path"] for item in frames)
            or any(not isinstance(item["display_id"], str) or not item["display_id"] for item in frames)
            or any(SHA256.fullmatch(str(item["screenshot_sha256"])) is None for item in frames)
        ):
            return emit("FAIL", "frame_population", manifest_sha256=manifest_sha, reason="frame identity or type invalid")
        frame_capture_epochs = [
            capture_epoch(item["captured_at"], "frame.captured_at")
            for item in frames
        ]
        if any(
            now - observed > float(policy["max_frame_age_seconds"])
            or observed - now > float(policy["max_future_skew_seconds"])
            for observed in frame_capture_epochs
        ):
            return emit("INSUFFICIENT_DATA", "frame_population", manifest_sha256=manifest_sha, reason="one or more frames are stale or future-dated")
        ordered = sorted(frames, key=lambda item: item.get("sequence", -1))
        if ordered != frames or [item.get("sequence") for item in frames] != list(range(len(frames))):
            return emit("FAIL", "frame_population", manifest_sha256=manifest_sha, reason="frame sequence invalid")
        population_projection = [
            {
                "frame_id": item.get("frame_id"), "sequence": item.get("sequence"),
                "screenshot_sha256": item.get("screenshot_sha256"),
                "size_bytes": item.get("size_bytes"), "display_id": item.get("display_id"),
                "captured_at": item.get("captured_at"),
            }
            for item in frames
        ]
        if (
            SHA256.fullmatch(str(manifest["measured_population_sha256"])) is None
            or sha256(canonical(population_projection)) != manifest["measured_population_sha256"]
        ):
            return emit("FAIL", "frame_population", manifest_sha256=manifest_sha, reason="population binding invalid")
        frame_bytes: list[bytes] = []
        ocr_matches: list[dict] = []
        compiled = [re.compile(pattern, re.IGNORECASE) for pattern in policy["prefilter_patterns"]]
        for item in frames:
            raw = read_frame(
                Path(item["path"]), int(item["size_bytes"]), item["screenshot_sha256"],
                int(policy["max_frame_bytes"]),
            )
            frame_bytes.append(raw)
            text = pytesseract.image_to_string(Image.open(io.BytesIO(raw)))
            matches = [pattern.pattern for pattern in compiled if pattern.search(text)]
            if matches:
                ocr_matches.append({"frame_id": item["frame_id"], "patterns": matches})
        if ocr_matches:
            return emit(
                "FAIL", "ocr_prefilter", policy_sha256=policy_sha,
                manifest_sha256=manifest_sha, ocr_matches=ocr_matches,
                measured_population_sha256=manifest["measured_population_sha256"],
                measured_population_count=len(frames),
            )
        request_id = str(uuid4())
        classifier_request = {
            "schema_version": "aidefend.desktop-classifier-request.v1",
            "request_id": request_id, "authenticated_actor_id": authenticated_actor_id,
            "tenant_id": manifest["tenant_id"], "workspace_id": manifest["workspace_id"],
            "task_id": manifest["task_id"], "action_id": manifest["action_id"],
            "action_type": action_type, "policy_sha256": policy_sha,
            "manifest_sha256": manifest_sha,
            "measured_population_sha256": manifest["measured_population_sha256"],
            "measured_population_count": len(frames),
            "frames": [
                {
                    "frame_id": item["frame_id"], "screenshot_sha256": item["screenshot_sha256"],
                    "content_b64": base64.b64encode(raw).decode("ascii"),
                }
                for item, raw in zip(frames, frame_bytes, strict=True)
            ],
        }
        classifier, classifier_sha = verify_envelope(
            post_signed(
                policy["classifier_endpoint"],
                classifier_request,
                float(policy["service_timeout_seconds"]),
                int(policy["service_response_max_bytes"]),
                int(policy["service_response_chunk_bytes"]),
            ),
            CLASSIFIER_KEY, "aidefend.desktop-classifier-receipt.v1",
        )
        if (
            classifier.get("request_id") != request_id
            or classifier.get("classifier_id") != policy["classifier_id"]
            or classifier.get("classifier_version") != policy["classifier_version"]
            or classifier.get("classifier_model_sha256") != policy["classifier_model_sha256"]
            or classifier.get("manifest_sha256") != manifest_sha
            or classifier.get("measured_population_sha256") != manifest["measured_population_sha256"]
            or classifier.get("measured_population_count") != len(frames)
            or classifier.get("population_complete") is not True
            or classifier.get("outcome") not in {"ALLOW", "BLOCK", "ABSTAIN"}
        ):
            return emit("FAIL", "classifier", classifier_receipt_sha256=classifier_sha, reason="classifier binding differs")
        scores = classifier.get("frame_scores")
        if (
            not isinstance(scores, list) or len(scores) != len(frames)
            or any(not isinstance(item, dict) or set(item) != {"frame_id", "injection_score"} for item in scores)
            or {item.get("frame_id") for item in scores} != {item["frame_id"] for item in frames}
            or len({item.get("frame_id") for item in scores}) != len(scores)
            or any(
                isinstance(item.get("injection_score"), bool)
                or not isinstance(item.get("injection_score"), (int, float))
                or not math.isfinite(float(item["injection_score"]))
                or not 0.0 <= float(item["injection_score"]) <= 1.0
                for item in scores
            )
        ):
            return emit("INSUFFICIENT_DATA", "classifier", classifier_receipt_sha256=classifier_sha, reason="classifier population incomplete")
        maximum = max(float(item["injection_score"]) for item in scores)
        if classifier.get("outcome") == "BLOCK":
            return emit(
                "FAIL", "classifier", classifier_receipt_sha256=classifier_sha,
                maximum_injection_score=maximum, reason="classifier returned a closed BLOCK outcome",
            )
        if classifier.get("outcome") == "ABSTAIN":
            needs_human = True
        elif maximum >= block_score:
            return emit(
                "FAIL", "classifier", classifier_receipt_sha256=classifier_sha,
                maximum_injection_score=maximum, reason="classifier block threshold reached",
            )
        else:
            needs_human = maximum > allow_score
        human_sha = None
        if needs_human:
            if policy.get("human_review_required_between_thresholds") is not True:
                return emit("INSUFFICIENT_DATA", "human_gate", classifier_receipt_sha256=classifier_sha, reason="policy has no bounded gray-zone route")
            review_request = {
                "schema_version": "aidefend.desktop-human-review-request.v1",
                "request_id": str(uuid4()), "authenticated_actor_id": authenticated_actor_id,
                "manifest_sha256": manifest_sha, "classifier_receipt_sha256": classifier_sha,
                "measured_population_sha256": manifest["measured_population_sha256"],
                "measured_population_count": len(frames), "action_type": action_type,
            }
            human, human_sha = verify_envelope(
                post_signed(
                    policy["human_review_endpoint"],
                    review_request,
                    float(policy["human_review_timeout_seconds"]),
                    int(policy["service_response_max_bytes"]),
                    int(policy["service_response_chunk_bytes"]),
                ),
                HUMAN_KEY, "aidefend.desktop-human-review-receipt.v1",
            )
            if (
                human.get("request_id") != review_request["request_id"]
                or human.get("manifest_sha256") != manifest_sha
                or human.get("classifier_receipt_sha256") != classifier_sha
                or human.get("measured_population_sha256") != manifest["measured_population_sha256"]
                or human.get("measured_population_count") != len(frames)
                or not human.get("reviewer_identity")
            ):
                return emit("FAIL", "human_gate", human_receipt_sha256=human_sha, reason="human receipt binding differs")
            if human.get("decision") == "DENY":
                return emit("FAIL", "human_gate", human_receipt_sha256=human_sha, reason="independent human denied")
            if human.get("decision") != "ALLOW":
                return emit("INSUFFICIENT_DATA", "human_gate", human_receipt_sha256=human_sha, reason="human review not final")
        require_watch_readback(manifest)
        return emit(
            "PASS", "completed", policy_sha256=policy_sha,
            manifest_sha256=manifest_sha, classifier_receipt_sha256=classifier_sha,
            human_receipt_sha256=human_sha,
            measured_population_sha256=manifest["measured_population_sha256"],
            measured_population_count=len(frames),
            maximum_injection_score=maximum,
        )
    except FileNotFoundError as exc:
        return emit("INSUFFICIENT_DATA", "gate", reason=type(exc).__name__)
    except (PermissionError, ValueError, BadSignatureError) as exc:
        return emit("FAIL", "gate", reason=type(exc).__name__)
    except requests.RequestException as exc:
        return emit("INSUFFICIENT_DATA", "gate", reason=type(exc).__name__)
    except Exception as exc:
        return emit("ERROR", "gate", reason=type(exc).__name__)

Require watch readiness for screenshot export

Install the G001 verify-only readiness module and verify the G005 worker's nonce-bound signed acknowledgement for the same workspace/task before the screenshot-risk decision and again before a completed allow result; the screenshot service needs no watchdog spool access, signing key or control/audit mTLS private key. A risk-gate N/A result alone does not authorize screen export; the trusted observation dispatcher must still satisfy watch readiness and every other applicable desktop boundary.

AID-I-008.004-G004

Platform scope, ownership, and assurance boundary

Use this method for a task-scoped disposable desktop/VDI workspace when a privileged observer can resolve one stable accessibility element and a separately credentialed readback service can measure the exact postcondition. AID-H-018.003 remains the only approval/gate-receipt issuer and AID-H-018.006 remains the generic execution-time anti-TOCTOU owner. This adapter supplies signed desktop observation evidence to that service; it never stores, signs, or decides approval itself. API-only actions are NOT_APPLICABLE only when the trusted dispatcher selects their API/tool path; model intent cannot select that state. Missing stable accessibility identity, exact expected invariant, complete observed population, or canonical service evidence is INSUFFICIENT_DATA and blocks invocation.

Signed invariant/readback policy example

{
  "schema_version": "aidefend.desktop-effect-policy.v1",
  "policy_version": "desktop-effects/2026.07",
  "policy_sha256": "sha256:POLICY_DIGEST",
  "derived_effect": "account_change",
  "target_resource": "account:user-2041",
  "invariant_id": "account-status-disabled/v3",
  "invariant_sha256": "sha256:INVARIANT_DIGEST",
  "expected_effect_sha256": "sha256:EXPECTED_CANONICAL_READBACK_DIGEST",
  "readback_source": "authoritative_api",
  "readback_adapter_id": "identity-admin-readback",
  "readback_adapter_version": "3.4.1",
  "complete_population_required": true
}

The trusted observer loads this signed policy outside the model workspace. For authoritative_api, its read-only mTLS adapter queries the exact resource after the UI invocation; for privileged_accessibility, it re-resolves the declared result element and hashes only that stable accessibility subtree and declared state fields, never the full DOM or screen. Bulk effects must enumerate the complete affected resource set and bind its count and digest.

Desktop-specific executor

# File: computer_use/exact_desktop_executor.py
from __future__ import annotations
import hashlib
import json
import re
from contextlib import AbstractContextManager
from dataclasses import asdict, dataclass
from typing import Final, Literal, Callable
from uuid import uuid4
from computer_use.desktop_clients import (
    ReadbackServiceError,
    ReadbackUnavailableError,
    connect_pinned_evidence_sink,
    connect_pinned_observer,
    request_independent_desktop_replay,
)
from h018.canonical_gate_adapter import (
    H018DeniedError,
    H018InsufficientDataError,
    H018ServiceError,
    consume_h018_receipt,
    request_h018_receipt,
    revoke_h018_receipt,
    verify_h018_receipt,
)
SHA256 = re.compile(r"sha256:[0-9a-f]{64}")
OUTCOMES = {"PASS", "FAIL", "INSUFFICIENT_DATA", "NOT_APPLICABLE", "ERROR"}
READBACK_SOURCES = {"authoritative_api", "privileged_accessibility"}
REPLAY_FIELDS = {
    "schema_version", "outcome", "action_id", "action_digest",
    "h018_receipt_sha256", "authorization_receipt_sha256",
    "invocation_receipt_sha256", "readback_receipt_sha256",
    "invariant_sha256", "measured_population_sha256",
    "verifier_identity", "receipt_sha256",
}
class DesktopActionBlocked(Exception):
    pass
@dataclass(frozen=True)
class GuiIntent:
    # Untrusted/model intent. No tenant, process, window, policy, effect, or applicability facts.
    task_id: str
    action_type: str
    target_resource: str
    requested_accessibility_element_id: str | None
@dataclass(frozen=True)
class PrivilegedUiBinding:
    tenant_id: str
    workspace_id: str
    session_id: str
    task_id: str
    foreground_process_identity: str
    target_application: str
    window_handle: str
    window_title: str
    accessibility_element_id: str
    accessibility_role: str
    accessibility_name: str
    accessibility_bounds: tuple[int, int, int, int]
    accessibility_subtree_sha256: str
    pre_action_state_sha256: str
    target_resource: str
    derived_effect: str
    execution_policy_version: str
    execution_policy_sha256: str
    invariant_id: str
    invariant_sha256: str
    expected_effect_sha256: str
    readback_source: Literal["authoritative_api", "privileged_accessibility"]
    readback_adapter_id: str
    readback_adapter_version: str
    complete_population_required: bool
@dataclass(frozen=True)
class InvocationReceipt:
    schema_version: str
    invocation_id: str
    action_id: str
    action_digest: str
    observer_identity: str
    invoked_at: str
    receipt_sha256: str
@dataclass(frozen=True)
class PostActionReadback:
    schema_version: str
    action_id: str
    action_digest: str
    outcome: Literal["PASS", "FAIL", "INSUFFICIENT_DATA", "ERROR"]
    reason: str
    observer_identity: str
    observed_at: str
    readback_source: Literal["authoritative_api", "privileged_accessibility"]
    readback_adapter_id: str
    readback_adapter_version: str
    target_resource: str
    invariant_id: str
    invariant_sha256: str
    expected_effect_sha256: str
    observed_effect_sha256: str | None
    observed_state_sha256: str | None
    measured_population_sha256: str | None
    measured_population_count: int | None
    population_complete: bool
    signed_readback_receipt_sha256: str
@dataclass(frozen=True)
class PrivilegedDesktopObserver:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    observe_target: Callable[[str | None], PrivilegedUiBinding]
    lock_exact_window: Callable[[str, str], AbstractContextManager[None]]
    activate_exact_window: Callable[[str, str], None]
    observe_locked_target: Callable[[str | None], PrivilegedUiBinding]
    invoke_stable_element: Callable[[str, str, PrivilegedUiBinding, str, dict], InvocationReceipt]
    readback_effect: Callable[[str, str, PrivilegedUiBinding, InvocationReceipt], PostActionReadback]
    freeze_workspace_and_revoke_leases: Callable[[str], None]
@dataclass(frozen=True)
class SignedEvidenceSink:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    append_and_verify: Callable[[dict], dict]
TRUSTED_OBSERVER: Final[PrivilegedDesktopObserver] = connect_pinned_observer(
    unix_socket="/run/aidefend/desktop-observer.sock",
    expected_peer_uid=0,
    expected_service_identity="aidefend-desktop-observer-prod",
)
TRUSTED_EVIDENCE: Final[SignedEvidenceSink] = connect_pinned_evidence_sink(
    unix_socket="/run/aidefend/desktop-observer.sock",
    expected_peer_uid=0,
    expected_service_identity="aidefend-desktop-observer-prod",
)
def canonical_sha256(value: dict) -> str:
    encoded = json.dumps(
        value, sort_keys=True, separators=(",", ":"), allow_nan=False
    ).encode("utf-8")
    return "sha256:" + hashlib.sha256(encoded).hexdigest()
def require_sha256(value: str, field: str) -> None:
    if not isinstance(value, str) or SHA256.fullmatch(value) is None:
        raise DesktopActionBlocked(field + " is not a sha256 binding")
def validate_binding(binding: PrivilegedUiBinding) -> None:
    required_strings = (
        "tenant_id", "workspace_id", "session_id", "task_id",
        "foreground_process_identity", "target_application", "window_handle",
        "window_title", "accessibility_element_id", "accessibility_role",
        "accessibility_name", "target_resource", "derived_effect",
        "execution_policy_version", "invariant_id", "readback_adapter_id",
        "readback_adapter_version",
    )
    if any(not isinstance(getattr(binding, field), str) or not getattr(binding, field)
           for field in required_strings):
        raise DesktopActionBlocked("privileged UI binding is incomplete")
    for field in (
        "accessibility_subtree_sha256", "pre_action_state_sha256",
        "execution_policy_sha256", "invariant_sha256", "expected_effect_sha256",
    ):
        require_sha256(getattr(binding, field), field)
    bounds = binding.accessibility_bounds
    if (
        not isinstance(bounds, tuple) or len(bounds) != 4
        or any(isinstance(item, bool) or not isinstance(item, int) for item in bounds)
        or bounds[2] <= 0 or bounds[3] <= 0
    ):
        raise DesktopActionBlocked("accessibility bounds are invalid")
    if binding.readback_source not in READBACK_SOURCES:
        raise DesktopActionBlocked("readback source is unsupported")
    if binding.complete_population_required is not True:
        raise DesktopActionBlocked("effect policy does not require complete population")
def validate_readback(
    readback: PostActionReadback,
    *,
    action_id: str,
    action_digest: str,
    binding: PrivilegedUiBinding,
) -> None:
    if (
        readback.schema_version != "aidefend.desktop-effect-readback.v1"
        or readback.action_id != action_id
        or readback.action_digest != action_digest
        or readback.outcome not in OUTCOMES - {"NOT_APPLICABLE"}
        or readback.readback_source != binding.readback_source
        or readback.readback_adapter_id != binding.readback_adapter_id
        or readback.readback_adapter_version != binding.readback_adapter_version
        or readback.target_resource != binding.target_resource
        or readback.invariant_id != binding.invariant_id
        or readback.invariant_sha256 != binding.invariant_sha256
        or readback.expected_effect_sha256 != binding.expected_effect_sha256
    ):
        raise DesktopActionBlocked("readback schema or binding differs")
    require_sha256(readback.signed_readback_receipt_sha256, "readback receipt")
    if readback.observed_effect_sha256 is not None:
        require_sha256(readback.observed_effect_sha256, "observed effect")
    if readback.observed_state_sha256 is not None:
        require_sha256(readback.observed_state_sha256, "observed state")
    if readback.measured_population_sha256 is not None:
        require_sha256(readback.measured_population_sha256, "measured population")
    if readback.outcome == "PASS" and (
        readback.observed_effect_sha256 != binding.expected_effect_sha256
        or readback.observed_state_sha256 is None
        or readback.measured_population_sha256 is None
        or not isinstance(readback.measured_population_count, int)
        or readback.measured_population_count < 1
        or readback.population_complete is not True
    ):
        raise DesktopActionBlocked("PASS readback lacks exact effect or complete population")
def validate_invocation(
    invocation: InvocationReceipt, *, action_id: str, action_digest: str
) -> None:
    if (
        invocation.schema_version != "aidefend.desktop-invocation-receipt.v1"
        or invocation.action_id != action_id
        or invocation.action_digest != action_digest
        or not invocation.invocation_id
        or not invocation.observer_identity
        or not invocation.invoked_at
    ):
        raise DesktopActionBlocked("invocation receipt schema or binding differs")
    require_sha256(invocation.receipt_sha256, "invocation receipt")
def h018_outcome(error: Exception) -> str:
    if isinstance(error, H018DeniedError):
        return "FAIL"
    if isinstance(error, H018InsufficientDataError):
        return "INSUFFICIENT_DATA"
    return "ERROR"
def emit(stage: str, outcome: str, **evidence: object) -> dict:
    if outcome not in OUTCOMES:
        raise ValueError("unknown outcome")
    return TRUSTED_EVIDENCE.append_and_verify({
        "schema_version": "aidefend.desktop-action-event.v1",
        "control": "AID-I-008.004",
        "event_id": str(uuid4()),
        "stage": stage,
        "outcome": outcome,
        **evidence,
    })
def execute_gui_action(intent: GuiIntent, authenticated_actor_id: str) -> dict:
    # Applicability is selected by the trusted dispatcher before this desktop-only
    # executor is called; untrusted intent cannot force NOT_APPLICABLE.
    if not authenticated_actor_id:
        raise DesktopActionBlocked("authenticated actor is required")
    observer = TRUSTED_OBSERVER
    action_id = str(uuid4())
    initial = observer.observe_target(intent.requested_accessibility_element_id)
    validate_binding(initial)
    if (
        intent.task_id != initial.task_id
        or intent.target_resource != initial.target_resource
        or intent.requested_accessibility_element_id != initial.accessibility_element_id
    ):
        emit(
            "pre_authorization", "FAIL", action_id=action_id,
            reason="intent differs from privileged desktop observation",
            intent=asdict(intent), privileged_binding=asdict(initial),
        )
        raise DesktopActionBlocked("intent and privileged observation differ")
    with observer.lock_exact_window(
        initial.window_handle, initial.foreground_process_identity
    ):
        observer.activate_exact_window(
            initial.window_handle, initial.foreground_process_identity
        )
        locked = observer.observe_locked_target(
            intent.requested_accessibility_element_id
        )
        validate_binding(locked)
        if locked != initial:
            emit(
                "pre_authorization", "FAIL", action_id=action_id,
                reason="desktop binding drifted before authorization",
                initial_binding=asdict(initial), locked_binding=asdict(locked),
            )
            raise DesktopActionBlocked("desktop binding drifted before authorization")
        bound_plan = {
            "schema_version": "aidefend.i008004-bound-desktop-action.v1",
            "action_id": action_id,
            "authenticated_actor_id": authenticated_actor_id,
            "intent": asdict(intent),
            "privileged_ui_binding": asdict(locked),
        }
        action_digest = canonical_sha256(bound_plan)
        observation_receipt = emit(
            "desktop_observation", "INSUFFICIENT_DATA",
            action_id=action_id, action_digest=action_digest,
            bound_plan=bound_plan,
            reason="signed observation pending canonical H-018 decision",
        )
        try:
            h018 = request_h018_receipt(
                action_id=action_id,
                action_digest=action_digest,
                policy_version=locked.execution_policy_version,
                policy_sha256=locked.execution_policy_sha256,
                authenticated_actor_id=authenticated_actor_id,
                additional_evidence_sha256s=[observation_receipt["receipt_sha256"]],
            )
            verify_h018_receipt(
                h018,
                action_id=action_id,
                action_digest=action_digest,
                policy_version=locked.execution_policy_version,
            )
        except (H018DeniedError, H018InsufficientDataError, H018ServiceError) as exc:
            emit(
                "h018_gate", h018_outcome(exc), action_id=action_id,
                action_digest=action_digest, reason=type(exc).__name__,
            )
            raise DesktopActionBlocked("canonical H-018 gate did not authorize") from exc
        final = observer.observe_locked_target(
            intent.requested_accessibility_element_id
        )
        validate_binding(final)
        if final != locked:
            revoke_h018_receipt(
                h018, action_id=action_id, action_digest=action_digest,
                reason="privileged_desktop_binding_drift",
            )
            emit(
                "pre_invocation", "FAIL", action_id=action_id,
                action_digest=action_digest,
                reason="desktop binding drifted after approval",
                locked_binding=asdict(locked), final_binding=asdict(final),
            )
            raise DesktopActionBlocked("desktop binding drifted after approval")
        try:
            consumed = consume_h018_receipt(
                h018,
                action_id=action_id,
                action_digest=action_digest,
                execution_binding_sha256=canonical_sha256(asdict(final)),
            )
        except (H018DeniedError, H018InsufficientDataError, H018ServiceError) as exc:
            emit(
                "h018_consume", h018_outcome(exc), action_id=action_id,
                action_digest=action_digest, reason=type(exc).__name__,
            )
            raise DesktopActionBlocked("canonical H-018 receipt was not consumed") from exc
        authorization_receipt = emit(
            "authorized", "PASS", action_id=action_id,
            action_digest=action_digest,
            h018_receipt_sha256=consumed["receipt_sha256"],
            privileged_ui_binding=asdict(final),
        )
        try:
            invocation = observer.invoke_stable_element(
                action_id=action_id,
                action_digest=action_digest,
                binding=final,
                action_type=intent.action_type,
                h018_consumption_receipt=consumed,
            )
            validate_invocation(
                invocation, action_id=action_id, action_digest=action_digest
            )
        except Exception as exc:
            observer.freeze_workspace_and_revoke_leases(reason="invocation_error")
            emit(
                "invocation", "ERROR", action_id=action_id,
                action_digest=action_digest,
                authorization_receipt_sha256=authorization_receipt["receipt_sha256"],
                reason=type(exc).__name__,
            )
            raise
        # An invocation receipt proves only that the stable element was invoked.
        # It is never interpreted as proof that the business/UI effect succeeded.
        try:
            readback = observer.readback_effect(
                action_id=action_id,
                action_digest=action_digest,
                binding=final,
                invocation=invocation,
            )
            validate_readback(
                readback, action_id=action_id,
                action_digest=action_digest, binding=final,
            )
        except ReadbackUnavailableError as exc:
            observer.freeze_workspace_and_revoke_leases(reason="readback_unavailable")
            receipt = emit(
                "effect_readback", "INSUFFICIENT_DATA", action_id=action_id,
                action_digest=action_digest, reason=type(exc).__name__,
                invocation_receipt_sha256=invocation.receipt_sha256,
            )
            return {
                "outcome": "INSUFFICIENT_DATA", "action_id": action_id,
                "evidence_receipt_sha256": receipt["receipt_sha256"],
            }
        except (ReadbackServiceError, DesktopActionBlocked) as exc:
            observer.freeze_workspace_and_revoke_leases(reason="readback_error")
            receipt = emit(
                "effect_readback", "ERROR", action_id=action_id,
                action_digest=action_digest, reason=type(exc).__name__,
                invocation_receipt_sha256=invocation.receipt_sha256,
            )
            return {
                "outcome": "ERROR", "action_id": action_id,
                "evidence_receipt_sha256": receipt["receipt_sha256"],
            }
        readback_event = emit(
            "effect_readback", readback.outcome,
            action_id=action_id, action_digest=action_digest,
            authorization_receipt_sha256=authorization_receipt["receipt_sha256"],
            invocation_receipt=asdict(invocation),
            post_action_readback=asdict(readback),
        )
        if readback.outcome != "PASS":
            observer.freeze_workspace_and_revoke_leases(
                reason="post_action_readback_" + readback.outcome.lower()
            )
            return {
                "outcome": readback.outcome,
                "action_id": action_id,
                "action_digest": action_digest,
                "readback_receipt_sha256": readback.signed_readback_receipt_sha256,
                "evidence_receipt_sha256": readback_event["receipt_sha256"],
            }
        replay = request_independent_desktop_replay(
            action_id=action_id,
            action_digest=action_digest,
            h018_receipt_sha256=consumed["receipt_sha256"],
            authorization_receipt_sha256=authorization_receipt["receipt_sha256"],
            invocation_receipt_sha256=invocation.receipt_sha256,
            readback_receipt_sha256=readback.signed_readback_receipt_sha256,
            expected_invariant_sha256=final.invariant_sha256,
            expected_population_sha256=readback.measured_population_sha256,
        )
        replay_expected = {
            "action_id": action_id,
            "action_digest": action_digest,
            "h018_receipt_sha256": consumed["receipt_sha256"],
            "authorization_receipt_sha256": authorization_receipt["receipt_sha256"],
            "invocation_receipt_sha256": invocation.receipt_sha256,
            "readback_receipt_sha256": readback.signed_readback_receipt_sha256,
            "invariant_sha256": final.invariant_sha256,
            "measured_population_sha256": readback.measured_population_sha256,
        }
        if (
            not isinstance(replay, dict)
            or set(replay) != REPLAY_FIELDS
            or replay.get("schema_version") != "aidefend.desktop-action-replay.v1"
            or any(replay.get(field) != value for field, value in replay_expected.items())
            or not isinstance(replay.get("verifier_identity"), str)
            or not replay.get("verifier_identity")
            or not isinstance(replay.get("receipt_sha256"), str)
            or SHA256.fullmatch(replay["receipt_sha256"]) is None
        ):
            observer.freeze_workspace_and_revoke_leases(reason="invalid_replay_receipt")
            emit(
                "independent_replay", "ERROR", action_id=action_id,
                action_digest=action_digest, reason="replay receipt binding differs",
            )
            return {"outcome": "ERROR", "action_id": action_id}
        if replay.get("outcome") != "PASS":
            observer.freeze_workspace_and_revoke_leases(reason="independent_replay_failed")
            emit(
                "independent_replay", "FAIL", action_id=action_id,
                action_digest=action_digest, replay_receipt=replay,
            )
            return {"outcome": "FAIL", "action_id": action_id}
        completion = emit(
            "completed", "PASS", action_id=action_id,
            action_digest=action_digest,
            h018_receipt_sha256=consumed["receipt_sha256"],
            invocation_receipt_sha256=invocation.receipt_sha256,
            readback_receipt_sha256=readback.signed_readback_receipt_sha256,
            independent_replay_receipt_sha256=replay["receipt_sha256"],
            observed_effect_sha256=readback.observed_effect_sha256,
            invariant_sha256=readback.invariant_sha256,
            measured_population_sha256=readback.measured_population_sha256,
            measured_population_count=readback.measured_population_count,
        )
        return {
            "outcome": "PASS",
            "action_id": action_id,
            "action_digest": action_digest,
            "completion_receipt_sha256": completion["receipt_sha256"],
        }

Concrete readback requirement: the pinned observer client accepts only a detached, signature-verified receipt from a readback service identity that has no GUI invocation or H-018 approval permission. Its versioned adapter must actually query the declared authoritative API resource after the action, or re-resolve the declared accessibility result subtree when no API exists. The receipt binds the exact target, adapter/version, invariant/policy digests, expected and observed effect digests, observation time, complete measured-population digest/count, and source. A return from invoke_stable_element, a changed screenshot, or an unverified model statement is never effect evidence.

Persist post-effect uncertainty before returning

Freezing the workspace is necessary but does not answer whether an invoked business effect occurred. Before invoke_stable_element(), conditionally reserve a durable action row keyed by action_id in PENDING_INVOCATION, binding the action digest, exact UI binding, consumed H-018 receipt, policy/invariant, and expected effect. After a valid invocation receipt, move it to EFFECT_UNCERTAIN before readback. Any readback, replay, evidence, or process failure after invocation keeps that row and the frozen/revoked workspace; it must not return an ordinary PASS, retry the GUI action, or discard the invocation receipt.

pending = reconciliation.reserve(
    action_id=action_id,
    state="PENDING_INVOCATION",
    action_digest=action_digest,
    ui_binding_sha256=canonical_sha256(asdict(final)),
    h018_receipt_sha256=consumed["receipt_sha256"],
    expected_effect_sha256=final.expected_effect_sha256,
)
invocation = observer.invoke_stable_element(
    action_id=action_id,
    action_digest=action_digest,
    binding=final,
    action_type=intent.action_type,
    h018_consumption_receipt=consumed,
)
validate_invocation(invocation, action_id=action_id, action_digest=action_digest)
reconciliation.compare_and_set(
    action_id, "PENDING_INVOCATION", "EFFECT_UNCERTAIN",
    invocation_receipt_sha256=invocation.receipt_sha256,
)

A separately credentialed reconciler reloads the exact row, calls only the declared authoritative API/accessibility readback adapter, validates the same invariant and complete measured population, and conditionally closes it as EFFECT_CONFIRMED, NO_EFFECT_CONFIRMED, or terminal FAIL. It never re-invokes the UI element. Include the reconciliation row/receipt in the final event; until closure, return INSUFFICIENT_DATA or ERROR and keep the workspace blocked.

AID-I-008.004-G005

Concept:

Computer-use sessions require an independently deployed watchdog with control-plane credentials unavailable to the model workspace. A terminate acknowledgement is only the start of containment: persist it durably, use a monotonic deadline to poll until a terminal receipt proves both workspace destruction and lease revocation, and keep the model session blocked throughout. Audit delivery is retried and durably spooled; an audit outage never returns control to the agent.

Watch-mode controller with terminal receipt

# File: computer_use/watch_mode.py
from __future__ import annotations

import base64
import hashlib
import json
import math
import os
import re
import stat
import time
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import quote

import jcs
import requests
from nacl.exceptions import BadSignatureError
from nacl.signing import SigningKey, VerifyKey


POLICY_PATH = Path("/etc/aidefend/watchdog-policy.signed.json")
SPOOL_PATH = Path("/var/lib/computer-use-watchdog/kill-switch-retry.jsonl")
OPAQUE_ID = re.compile(r"[A-Za-z0-9._:-]{1,128}")
SHA256 = re.compile(r"sha256:[0-9a-f]{64}")
OUTCOMES = {"PASS", "FAIL", "INSUFFICIENT_DATA", "NOT_APPLICABLE", "ERROR"}
POLICY_VERIFY_KEY = VerifyKey(bytes.fromhex(os.environ["WATCHDOG_POLICY_VERIFY_KEY_HEX"]))
CONTROL_VERIFY_KEY = VerifyKey(bytes.fromhex(os.environ["WORKSPACE_CONTROL_VERIFY_KEY_HEX"]))
AUDIT_VERIFY_KEY = VerifyKey(bytes.fromhex(os.environ["EXTERNAL_AUDIT_VERIFY_KEY_HEX"]))
TELEMETRY_VERIFY_KEY = VerifyKey(bytes.fromhex(os.environ["WATCHDOG_TELEMETRY_VERIFY_KEY_HEX"]))
EVIDENCE_KEY = SigningKey(bytes.fromhex(os.environ["WATCHDOG_EVIDENCE_SIGNING_KEY_HEX"]))
CONTROL_CERT = (os.environ["WATCHDOG_CONTROL_CERT"], os.environ["WATCHDOG_CONTROL_KEY"])
AUDIT_CERT = (os.environ["WATCHDOG_AUDIT_CERT"], os.environ["WATCHDOG_AUDIT_KEY"])
CA_BUNDLE = os.environ["INTERNAL_CA_BUNDLE"]
RUNTIME_PROFILE_VERSION = os.environ["AIDEFEND_RUNTIME_PROFILE_VERSION"]
RUNTIME_PROFILE_SHA256 = os.environ["AIDEFEND_RUNTIME_PROFILE_SHA256"]
POLICY_MIN_BYTES = int(os.environ["WATCHDOG_POLICY_MIN_BYTES"])
POLICY_MAX_BYTES = int(os.environ["WATCHDOG_POLICY_MAX_BYTES"])
if (
    not RUNTIME_PROFILE_VERSION
    or re.fullmatch(r"[0-9a-f]{64}", RUNTIME_PROFILE_SHA256) is None
    or POLICY_MIN_BYTES < 1
    or POLICY_MAX_BYTES < POLICY_MIN_BYTES
):
    raise RuntimeError("verified watchdog bootstrap profile is invalid")


@dataclass(frozen=True)
class WatchEvent:
    event_type: str
    policy_version: str
    policy_sha256: str
    runtime_profile_version: str
    runtime_profile_sha256: str
    task_id: str
    workspace_id: str
    reason: str
    requested_at: str
    completed_at: str
    termination_receipt_id: str
    terminal_receipt_sha256: str
    workspace_state: str
    lease_state: str
    disposable_storage_destroyed: bool


class KillSwitchTriggered(Exception):
    """The calling session must remain blocked and may not resume model work."""


def canonical(value: object) -> bytes:
    try:
        return jcs.canonicalize(value)
    except (RecursionError, OverflowError) as exc:
        raise ValueError("watchdog JSON exceeds supported depth or numeric range") from exc


def canonical_sha256(value: object) -> str:
    return "sha256:" + hashlib.sha256(canonical(value)).hexdigest()


def strict_json_loads(raw: bytes | str) -> object:
    def reject_constant(value: str) -> None:
        raise ValueError("non-standard JSON constant: " + value)

    def unique_object(pairs: list[tuple[str, object]]) -> dict:
        result = {}
        for key, value in pairs:
            if key in result:
                raise ValueError("duplicate JSON key: " + key)
            result[key] = value
        return result

    try:
        return json.loads(raw, parse_constant=reject_constant, object_pairs_hook=unique_object)
    except (RecursionError, OverflowError) as exc:
        raise ValueError("watchdog JSON exceeds supported depth or numeric range") from exc


def bounded_response_json(response: requests.Response) -> object:
    declared = response.headers.get("content-length")
    if declared is not None:
        try:
            declared_size = int(declared)
        except ValueError as exc:
            raise ValueError("invalid watchdog response content-length") from exc
        if declared_size < 0 or declared_size > HTTP_RESPONSE_MAX_BYTES:
            raise ValueError("watchdog response exceeds signed policy")
    raw = bytearray()
    for chunk in response.iter_content(chunk_size=HTTP_RESPONSE_CHUNK_BYTES):
        raw.extend(chunk)
        if len(raw) > HTTP_RESPONSE_MAX_BYTES:
            raise ValueError("watchdog response exceeds signed policy")
    if not raw:
        raise ValueError("watchdog response is empty")
    return strict_json_loads(bytes(raw))


def verify_envelope(value: object, key: VerifyKey, schema: str) -> tuple[dict, str]:
    if not isinstance(value, dict) or set(value) != {"payload", "signature_b64"}:
        raise ValueError("signed receipt envelope schema differs")
    payload = value["payload"]
    if not isinstance(payload, dict) or payload.get("schema_version") != schema:
        raise ValueError("signed receipt payload schema differs")
    signature = value["signature_b64"]
    if not isinstance(signature, str) or not signature:
        raise ValueError("signed receipt signature must be a nonempty base64 string")
    body = canonical(payload)
    key.verify(body, base64.b64decode(signature, validate=True))
    return payload, "sha256:" + hashlib.sha256(body).hexdigest()


def load_policy() -> tuple[dict, str]:
    descriptor = os.open(POLICY_PATH, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
    try:
        info = os.fstat(descriptor)
        if not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o022:
            raise PermissionError("watchdog policy ownership/mode invalid")
        if info.st_size < POLICY_MIN_BYTES or info.st_size > POLICY_MAX_BYTES:
            raise ValueError("watchdog policy size invalid")
        raw = os.read(descriptor, info.st_size + 1)
        if len(raw) != info.st_size:
            raise OSError("watchdog policy changed during read")
    finally:
        os.close(descriptor)
    policy, policy_sha = verify_envelope(
        strict_json_loads(raw), POLICY_VERIFY_KEY, "aidefend.desktop-watchdog-policy.v2"
    )
    fields = {
        "schema_version", "policy_version", "valid_from_unix", "valid_until_unix",
        "workspace_control_url", "external_audit_url", "telemetry_endpoint",
        "max_task_seconds", "max_policy_denials", "termination_timeout_seconds",
        "audit_retry_seconds", "poll_interval_seconds", "http_timeout_seconds",
        "telemetry_max_age_seconds", "telemetry_max_future_skew_seconds",
        "audit_backoff_initial_seconds", "audit_backoff_multiplier",
        "audit_backoff_max_seconds",
        "http_response_max_bytes", "http_response_chunk_bytes",
        "collector_identity", "collector_generation", "readback_client_uids", "readback_timeout_seconds",
    }
    if set(policy) != fields:
        raise ValueError("watchdog policy fields differ")
    for field in ("valid_from_unix", "valid_until_unix"):
        value = policy[field]
        if (
            isinstance(value, bool)
            or not isinstance(value, (int, float))
            or not math.isfinite(float(value))
        ):
            raise ValueError("watchdog policy validity bound is invalid")
    if (
        not isinstance(policy["collector_identity"], str)
        or OPAQUE_ID.fullmatch(policy["collector_identity"]) is None
        or isinstance(policy["collector_generation"], bool)
        or not isinstance(policy["collector_generation"], int)
        or policy["collector_generation"] < 1
    ):
        raise ValueError("watchdog collector binding is invalid")
    uids = policy["readback_client_uids"]
    if (not isinstance(uids, list) or not uids
            or any(type(uid) is not int or uid < 1 for uid in uids)
            or len(set(uids)) != len(uids)):
        raise ValueError("watchdog readback client identities are invalid")
    now = time.time()
    if not float(policy["valid_from_unix"]) <= now <= float(policy["valid_until_unix"]):
        raise PermissionError("watchdog policy is not currently valid")
    if any(not str(policy[field]).startswith("https://") for field in (
        "workspace_control_url", "external_audit_url", "telemetry_endpoint"
    )):
        raise ValueError("watchdog endpoints must use HTTPS")
    positive_fields = {
        "max_task_seconds", "termination_timeout_seconds", "audit_retry_seconds", "readback_timeout_seconds",
        "poll_interval_seconds", "http_timeout_seconds",
        "telemetry_max_age_seconds", "audit_backoff_initial_seconds",
        "audit_backoff_multiplier", "audit_backoff_max_seconds",
    }
    for field in positive_fields:
        value = policy[field]
        if (
            isinstance(value, bool)
            or not isinstance(value, (int, float))
            or not math.isfinite(float(value))
            or float(value) <= 0
        ):
            raise ValueError("watchdog signed positive bound invalid: " + field)
    denial_limit = policy["max_policy_denials"]
    if isinstance(denial_limit, bool) or not isinstance(denial_limit, int) or denial_limit < 1:
        raise ValueError("watchdog signed denial limit is invalid")
    future_skew = policy["telemetry_max_future_skew_seconds"]
    if (
        isinstance(future_skew, bool)
        or not isinstance(future_skew, (int, float))
        or not math.isfinite(float(future_skew))
        or float(future_skew) < 0
    ):
        raise ValueError("watchdog signed future-skew bound is invalid")
    if float(policy["audit_backoff_max_seconds"]) < float(policy["audit_backoff_initial_seconds"]):
        raise ValueError("watchdog signed audit backoff range is inverted")
    for field in ("http_response_max_bytes", "http_response_chunk_bytes"):
        value = policy[field]
        if isinstance(value, bool) or not isinstance(value, int) or value < 1:
            raise ValueError("watchdog signed response bound invalid: " + field)
    if (
        policy["http_response_max_bytes"] < 2
        or policy["http_response_chunk_bytes"] > policy["http_response_max_bytes"]
    ):
        raise ValueError("watchdog signed response bounds are inconsistent")
    return policy, policy_sha


POLICY, POLICY_SHA256 = load_policy()
WORKSPACE_CONTROL_URL = str(POLICY["workspace_control_url"]).rstrip("/")
EXTERNAL_AUDIT_URL = str(POLICY["external_audit_url"])
TERMINATION_TIMEOUT_SECONDS = float(POLICY["termination_timeout_seconds"])
AUDIT_RETRY_SECONDS = float(POLICY["audit_retry_seconds"])
POLL_INTERVAL_SECONDS = float(POLICY["poll_interval_seconds"])
HTTP_TIMEOUT_SECONDS = float(POLICY["http_timeout_seconds"])
TELEMETRY_MAX_AGE_SECONDS = float(POLICY["telemetry_max_age_seconds"])
TELEMETRY_MAX_FUTURE_SKEW_SECONDS = float(POLICY["telemetry_max_future_skew_seconds"])
AUDIT_BACKOFF_INITIAL_SECONDS = float(POLICY["audit_backoff_initial_seconds"])
AUDIT_BACKOFF_MULTIPLIER = float(POLICY["audit_backoff_multiplier"])
AUDIT_BACKOFF_MAX_SECONDS = float(POLICY["audit_backoff_max_seconds"])
HTTP_RESPONSE_MAX_BYTES = int(POLICY["http_response_max_bytes"])
HTTP_RESPONSE_CHUNK_BYTES = int(POLICY["http_response_chunk_bytes"])


def require_current_loaded_policy() -> None:
    # Re-open, re-verify, and re-check time bounds for every sample and
    # immediately before PASS. Import-time validity is not currentness.
    current, current_sha256 = load_policy()
    if current_sha256 != POLICY_SHA256 or current != POLICY:
        raise PermissionError("watchdog policy rotated; restart against the new generation")


def persist_spool(stage: str, outcome: str, payload: dict) -> dict:
    # Root-owned O_APPEND file is the local non-overwrite queue; external audit
    # and its independently keyed receipt remain the audit authority.
    try:
        if outcome not in OUTCOMES:
            raise ValueError("invalid watchdog outcome")
        SPOOL_PATH.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        record = {
            "schema_version": "aidefend.computer_use_watch_spool.v1",
            "control": "AID-I-008.004", "stage": stage, "outcome": outcome,
            "record_id": str(__import__("uuid").uuid4()),
            "policy_version": POLICY["policy_version"],
            "policy_sha256": POLICY_SHA256,
            "runtime_profile_version": RUNTIME_PROFILE_VERSION,
            "runtime_profile_sha256": RUNTIME_PROFILE_SHA256,
            "spooled_at": datetime.now(timezone.utc).isoformat(),
            "process_monotonic_ns": time.monotonic_ns(),
            "payload": payload,
        }
        body = canonical(record)
        signature = EVIDENCE_KEY.sign(body).signature
        envelope = {
            "record": record, "record_sha256": canonical_sha256(record),
            "signature_b64": base64.b64encode(signature).decode("ascii"),
        }
        line = canonical(envelope) + b"\n"
        descriptor = os.open(
            SPOOL_PATH,
            os.O_APPEND | os.O_CREAT | os.O_RDWR | os.O_CLOEXEC | os.O_NOFOLLOW,
            0o600,
        )
        try:
            info = os.fstat(descriptor)
            if not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o077:
                raise PermissionError("watchdog spool ownership/mode invalid")
            offset = os.lseek(descriptor, 0, os.SEEK_END)
            if os.write(descriptor, line) != len(line):
                raise OSError("short write to watchdog spool")
            os.fsync(descriptor)
            readback = os.pread(descriptor, len(line), offset)
        finally:
            os.close(descriptor)
        if readback != line:
            raise OSError("watchdog spool durable readback differs")
        verified = strict_json_loads(readback)
        if verified != envelope:
            raise OSError("watchdog spool parsed readback differs")
        EVIDENCE_KEY.verify_key.verify(body, base64.b64decode(verified["signature_b64"], validate=True))
        return envelope
    except (OSError, ValueError, PermissionError) as exc:
        raise KillSwitchTriggered(
            "durable watchdog spool unavailable; session remains blocked"
        ) from exc


def validate_context(task_id: str, workspace_id: str, reason: str) -> None:
    for name, value in {
        "task_id": task_id,
        "workspace_id": workspace_id,
        "reason": reason,
    }.items():
        if not isinstance(value, str) or OPAQUE_ID.fullmatch(value) is None:
            raise ValueError(f"invalid {name}")


def poll_terminal_receipt(
    *,
    task_id: str,
    workspace_id: str,
    termination_receipt_id: str,
    deadline: float,
) -> dict:
    receipt_url = (
        f"{WORKSPACE_CONTROL_URL}/v1/termination-receipts/"
        f"{quote(termination_receipt_id, safe='')}"
    )
    last_error = "no terminal receipt observed"
    while time.monotonic() < deadline:
        try:
            with requests.get(
                receipt_url,
                cert=CONTROL_CERT,
                verify=CA_BUNDLE,
                timeout=HTTP_TIMEOUT_SECONDS,
                allow_redirects=False,
                stream=True,
            ) as response:
                response.raise_for_status()
                value = bounded_response_json(response)
            receipt, receipt_sha256 = verify_envelope(
                value, CONTROL_VERIFY_KEY,
                "aidefend.workspace-termination-receipt.v1",
            )
            if (
                receipt.get("termination_receipt_id") != termination_receipt_id
                or receipt.get("task_id") != task_id
                or receipt.get("workspace_id") != workspace_id
            ):
                raise RuntimeError("termination receipt context mismatch")
            if (
                receipt.get("state") == "destroyed"
                and receipt.get("lease_state") == "revoked"
                and receipt.get("disposable_storage_destroyed") is True
            ):
                return {**receipt, "receipt_sha256": receipt_sha256}
            if receipt.get("state") not in {"requested", "terminating", "destroying"}:
                raise RuntimeError("unexpected non-terminal workspace state")
            last_error = f"workspace still {receipt.get('state')}"
        except (requests.RequestException, ValueError, RuntimeError, BadSignatureError) as exc:
            last_error = type(exc).__name__
        remaining = deadline - time.monotonic()
        if remaining > 0:
            time.sleep(min(POLL_INTERVAL_SECONDS, remaining))
    raise TimeoutError(f"terminal destruction receipt not observed: {last_error}")


def deliver_external_audit(event: WatchEvent, event_sha256: str) -> dict | None:
    deadline = time.monotonic() + AUDIT_RETRY_SECONDS
    attempt = 0
    while time.monotonic() < deadline:
        attempt += 1
        try:
            with requests.post(
                EXTERNAL_AUDIT_URL,
                json=asdict(event),
                headers={
                    "Idempotency-Key": event.termination_receipt_id,
                    "X-Event-SHA256": event_sha256,
                },
                cert=AUDIT_CERT,
                verify=CA_BUNDLE,
                timeout=HTTP_TIMEOUT_SECONDS,
                allow_redirects=False,
                stream=True,
            ) as response:
                response.raise_for_status()
                value = bounded_response_json(response)
            audit_receipt, audit_receipt_sha256 = verify_envelope(
                value, AUDIT_VERIFY_KEY,
                "aidefend.external-audit-receipt.v1",
            )
            if (
                audit_receipt.get("event_sha256") != event_sha256
                or not audit_receipt.get("audit_receipt_id")
            ):
                raise RuntimeError("audit service did not acknowledge exact event")
            return {**audit_receipt, "receipt_sha256": audit_receipt_sha256}
        except (requests.RequestException, ValueError, RuntimeError, BadSignatureError):
            remaining = deadline - time.monotonic()
            if remaining > 0:
                time.sleep(min(
                    AUDIT_BACKOFF_MAX_SECONDS,
                    AUDIT_BACKOFF_INITIAL_SECONDS
                    * AUDIT_BACKOFF_MULTIPLIER ** (attempt - 1),
                    remaining,
                ))
    return None


def trigger_kill_switch(task_id: str, workspace_id: str, reason: str) -> None:
    validate_context(task_id, workspace_id, reason)
    requested_at = datetime.now(timezone.utc).isoformat()
    idempotency_key = "kill:" + canonical_sha256({
        "task_id": task_id,
        "workspace_id": workspace_id,
        "reason": reason,
    })
    try:
        with requests.post(
            f"{WORKSPACE_CONTROL_URL}/v1/workspaces/{quote(workspace_id, safe='')}/terminate",
            json={
                "task_id": task_id,
                "reason": reason,
                "destroy_disposable_storage": True,
                "revoke_workspace_lease": True,
            },
            headers={"Idempotency-Key": idempotency_key},
            cert=CONTROL_CERT,
            verify=CA_BUNDLE,
            timeout=HTTP_TIMEOUT_SECONDS,
            allow_redirects=False,
            stream=True,
        ) as termination:
            termination.raise_for_status()
            value = bounded_response_json(termination)
        acknowledgement, acknowledgement_sha256 = verify_envelope(
            value, CONTROL_VERIFY_KEY,
            "aidefend.workspace-termination-ack.v1",
        )
        if (
            acknowledgement.get("task_id") != task_id
            or acknowledgement.get("workspace_id") != workspace_id
            or acknowledgement.get("state") not in {"requested", "terminating"}
            or not acknowledgement.get("termination_receipt_id")
        ):
            raise RuntimeError("workspace control plane acknowledgement invalid")
    except (requests.RequestException, ValueError, RuntimeError, BadSignatureError) as exc:
        try:
            persist_spool("termination_request_retry", "ERROR", {
                "task_id": task_id,
                "workspace_id": workspace_id,
                "reason": reason,
                "idempotency_key": idempotency_key,
                "error_type": type(exc).__name__,
            })
        finally:
            raise KillSwitchTriggered(
                "termination request is pending watchdog retry; session remains blocked"
            ) from exc

    termination_receipt_id = str(acknowledgement["termination_receipt_id"])
    if OPAQUE_ID.fullmatch(termination_receipt_id) is None:
        raise KillSwitchTriggered("invalid termination receipt; session remains blocked")
    persist_spool("termination_acknowledged", "INSUFFICIENT_DATA", {
        "task_id": task_id,
        "workspace_id": workspace_id,
        "reason": reason,
        "idempotency_key": idempotency_key,
        "acknowledgement": acknowledgement,
        "acknowledgement_sha256": acknowledgement_sha256,
    })

    deadline = time.monotonic() + TERMINATION_TIMEOUT_SECONDS
    try:
        terminal_receipt = poll_terminal_receipt(
            task_id=task_id,
            workspace_id=workspace_id,
            termination_receipt_id=termination_receipt_id,
            deadline=deadline,
        )
    except TimeoutError as exc:
        try:
            persist_spool("terminal_receipt_poll_retry", "INSUFFICIENT_DATA", {
                "task_id": task_id,
                "workspace_id": workspace_id,
                "reason": reason,
                "termination_receipt_id": termination_receipt_id,
            })
        finally:
            raise KillSwitchTriggered(
                "destruction not yet proven; session remains blocked and poll is queued"
            ) from exc

    terminal_receipt_sha256 = str(terminal_receipt["receipt_sha256"])
    persist_spool("termination_terminal", "FAIL", {
        "terminal_receipt": terminal_receipt,
        "terminal_receipt_sha256": terminal_receipt_sha256,
    })
    event = WatchEvent(
        event_type="computer_use_kill_switch_terminal",
        policy_version=POLICY["policy_version"],
        policy_sha256=POLICY_SHA256,
        runtime_profile_version=RUNTIME_PROFILE_VERSION,
        runtime_profile_sha256=RUNTIME_PROFILE_SHA256,
        task_id=task_id,
        workspace_id=workspace_id,
        reason=reason,
        requested_at=requested_at,
        completed_at=datetime.now(timezone.utc).isoformat(),
        termination_receipt_id=termination_receipt_id,
        terminal_receipt_sha256=terminal_receipt_sha256,
        workspace_state="destroyed",
        lease_state="revoked",
        disposable_storage_destroyed=True,
    )
    event_sha256 = canonical_sha256(asdict(event))
    audit_receipt = deliver_external_audit(event, event_sha256)
    if audit_receipt is None:
        # A separate retry worker replays this exact event. Crucially, the model
        # session still receives KillSwitchTriggered and cannot continue.
        persist_spool("audit_delivery_retry", "INSUFFICIENT_DATA", {
            "event": asdict(event),
            "event_sha256": event_sha256,
        })
        raise KillSwitchTriggered(
            "workspace destroyed; external audit queued and session remains blocked"
        )

    persist_spool("audit_delivered", "FAIL", {
        "event_sha256": event_sha256,
        "audit_receipt": audit_receipt,
    })
    raise KillSwitchTriggered("workspace destroyed, lease revoked, and audit recorded")


def enforce_watchdog(
    task_id: str | None,
    workspace_id: str | None,
    *, challenge_nonce: str | None = None,
) -> dict:
    if task_id is None and workspace_id is None:
        return persist_spool(
            "applicability", "NOT_APPLICABLE",
            {"reason": "no active disposable desktop workspace"},
        )
    if task_id is None or workspace_id is None:
        return persist_spool(
            "watch_sample", "INSUFFICIENT_DATA",
            {"reason": "task/workspace identity incomplete"},
        )
    validate_context(task_id, workspace_id, "watch_sample")
    if challenge_nonce is not None and re.fullmatch(r"[a-f0-9]{64}", challenge_nonce) is None:
        raise ValueError("watchdog readback challenge is invalid")
    endpoint = (
        str(POLICY["telemetry_endpoint"]).rstrip("/") + "/"
        + quote(workspace_id, safe="")
    )
    try:
        require_current_loaded_policy()
        with requests.get(
            endpoint, params={"task_id": task_id}, cert=CONTROL_CERT,
            verify=CA_BUNDLE, timeout=HTTP_TIMEOUT_SECONDS,
            allow_redirects=False, stream=True,
        ) as response:
            response.raise_for_status()
            value = bounded_response_json(response)
        telemetry, telemetry_sha256 = verify_envelope(
            value, TELEMETRY_VERIFY_KEY,
            "aidefend.desktop-watch-telemetry.v1",
        )
        fields = {
            "schema_version", "task_id", "workspace_id", "workspace_state",
            "elapsed_seconds", "policy_denials", "population_complete",
            "measured_population_count", "measured_population_sha256",
            "collector_identity", "collector_generation", "observed_at_unix",
        }
        if (
            set(telemetry) != fields
            or telemetry["task_id"] != task_id
            or telemetry["workspace_id"] != workspace_id
            or telemetry["workspace_state"] != "RUNNING"
            or telemetry["population_complete"] is not True
            or not isinstance(telemetry["collector_identity"], str)
            or isinstance(telemetry["collector_generation"], bool)
            or not isinstance(telemetry["collector_generation"], int)
            or isinstance(telemetry["measured_population_count"], bool)
            or not isinstance(telemetry["measured_population_count"], int)
            or telemetry["measured_population_count"] < 1
            or SHA256.fullmatch(str(telemetry["measured_population_sha256"])) is None
            or telemetry["collector_identity"] != POLICY["collector_identity"]
            or telemetry["collector_generation"] != POLICY["collector_generation"]
        ):
            return persist_spool(
                "watch_sample", "INSUFFICIENT_DATA",
                {
                    "task_id": task_id, "workspace_id": workspace_id,
                    "telemetry_receipt_sha256": telemetry_sha256,
                    "reason": "telemetry binding, freshness, or population incomplete",
                },
            )
        observed_at = telemetry["observed_at_unix"]
        elapsed_value = telemetry["elapsed_seconds"]
        denials_value = telemetry["policy_denials"]
        if (
            isinstance(observed_at, bool)
            or not isinstance(observed_at, (int, float))
            or not math.isfinite(float(observed_at))
            or isinstance(elapsed_value, bool)
            or not isinstance(elapsed_value, (int, float))
            or not math.isfinite(float(elapsed_value))
            or float(elapsed_value) < 0
            or isinstance(denials_value, bool)
            or not isinstance(denials_value, int)
            or denials_value < 0
        ):
            raise ValueError("watchdog telemetry numeric domain invalid")
        age = time.time() - float(observed_at)
        if age > TELEMETRY_MAX_AGE_SECONDS or age < -TELEMETRY_MAX_FUTURE_SKEW_SECONDS:
            return persist_spool(
                "watch_sample", "INSUFFICIENT_DATA",
                {
                    "task_id": task_id, "workspace_id": workspace_id,
                    "telemetry_receipt_sha256": telemetry_sha256,
                    "collector_identity": telemetry["collector_identity"],
                    "collector_generation": telemetry["collector_generation"],
                    "reason": "telemetry freshness outside signed policy",
                },
            )
        elapsed = float(elapsed_value)
        denials = denials_value
        if elapsed > float(POLICY["max_task_seconds"]):
            persist_spool(
                "watch_finding", "FAIL",
                {
                    "task_id": task_id, "workspace_id": workspace_id,
                    "telemetry_receipt_sha256": telemetry_sha256,
                    "reason": "task_time_limit_exceeded",
                },
            )
            trigger_kill_switch(task_id, workspace_id, "task_time_limit_exceeded")
        if denials >= int(POLICY["max_policy_denials"]):
            persist_spool(
                "watch_finding", "FAIL",
                {
                    "task_id": task_id, "workspace_id": workspace_id,
                    "telemetry_receipt_sha256": telemetry_sha256,
                    "reason": "repeated_policy_denials",
                },
            )
            trigger_kill_switch(task_id, workspace_id, "repeated_policy_denials")
        require_current_loaded_policy()
        return persist_spool(
            "watch_sample", "PASS",
            {
                "task_id": task_id, "workspace_id": workspace_id,
                "telemetry_receipt_sha256": telemetry_sha256,
                "collector_identity": telemetry["collector_identity"],
                "collector_generation": telemetry["collector_generation"],
                "measured_population_sha256": telemetry["measured_population_sha256"],
                "measured_population_count": telemetry["measured_population_count"],
                "elapsed_seconds": elapsed, "policy_denials": denials, "challenge_nonce": challenge_nonce,
            },
        )
    except (BadSignatureError, ValueError, PermissionError) as exc:
        persist_spool(
            "watch_sample", "FAIL",
            {"task_id": task_id, "workspace_id": workspace_id, "reason": type(exc).__name__},
        )
        trigger_kill_switch(task_id, workspace_id, "invalid_watch_telemetry")
        raise AssertionError("kill switch must not return")
    except requests.RequestException as exc:
        persist_spool(
            "watch_sample", "ERROR",
            {"task_id": task_id, "workspace_id": workspace_id, "reason": type(exc).__name__},
        )
        trigger_kill_switch(task_id, workspace_id, "watch_telemetry_unavailable")
        raise AssertionError("kill switch must not return")

Cleanup, evidence, and replay requirements

• Run the watchdog, telemetry collector, and spool replay worker outside the model workspace; none accepts model-supplied thresholds, endpoints, credentials, elapsed time, denial counts, or scope.
• Persist each signed local outcome before acting. Verify the control-plane and external-audit signatures plus exact task/workspace/receipt bindings; HTTP success alone is not evidence.
• Persist the termination acknowledgement before polling, and accept only a signed terminal receipt with state=destroyed, lease_state=revoked, and disposable-storage destruction.
• Use the short renewable workspace lease as a backstop and replay queued termination/audit records with their original idempotency keys. A read-only verifier must retrieve the exact spool record from a separate replica, verify its local signature, and compare the external receipt hashes before closure.
• Move only approved outputs to release storage; quarantine everything else and clear task-scoped clipboard/session state.

Expose the privileged worker through a local readback socket

Install the following companion to watch_mode.py and run it as the root-owned watchdog service. Provision /run/computer-use-watchdog with root ownership, observer-service group, mode 0750, and a root-controlled lifecycle; set WATCHDOG_READBACK_GROUP_ID from the protected deployment profile. The socket is root-owned mode 0660, and the signed v2 watchdog policy lists the non-root observer/screenshot readback_client_uids and a positive readback_timeout_seconds. Choose the latter to bound the complete local request, telemetry verification and durable acknowledgement rather than a single HTTP call. Only this worker holds the watchdog evidence signing key, watchdog control/audit mTLS private keys and root-owned 0600 spool. Provision the independently pinned readback verification public key and policy/runtime bootstrap for both worker and clients. For this watchdog interface the observer services hold public trust only; their own evidence-signing/control credentials remain separate, and the model workspace has no socket group/UID admission.

# File: computer_use/watch_readback_server.py
from __future__ import annotations
import os
import logging
import re
import socket
import stat
import time
from computer_use import watch_mode as watchdog
from computer_use.desktop_watch_readiness import SOCKET_PATH, peer_uid, read_frame, write_frame


LOGGER = logging.getLogger(__name__)


def handle_connection(stream):
    watchdog.require_current_loaded_policy()
    if peer_uid(stream) not in watchdog.POLICY["readback_client_uids"]:
        raise PermissionError("watchdog readback caller is not an authorized observer service")
    deadline = time.monotonic() + watchdog.POLICY["readback_timeout_seconds"]
    maximum = watchdog.HTTP_RESPONSE_MAX_BYTES
    request = read_frame(stream, maximum, deadline)
    fields = {"schema_version", "task_id", "workspace_id", "challenge_nonce"}
    if (not isinstance(request, dict) or set(request) != fields
            or request["schema_version"] != "aidefend.desktop-watch-readback-request.v1"
            or not isinstance(request["challenge_nonce"], str)
            or re.fullmatch(r"[a-f0-9]{64}", request["challenge_nonce"]) is None):
        raise ValueError("watchdog readback request schema differs")
    watchdog.validate_context(request["task_id"], request["workspace_id"], "watch_sample")
    envelope = watchdog.enforce_watchdog(request["task_id"], request["workspace_id"],
                                         challenge_nonce=request["challenge_nonce"])
    # enforce_watchdog returns only after the root-owned spool write, fsync,
    # exact readback and signature verification succeed. The observer has no
    # spool descriptor or signing key. Failure closes RPC and keeps it blocked.
    write_frame(stream, envelope, maximum, deadline)


def serve():
    if os.geteuid() != 0: raise PermissionError("watch readback worker must run as the privileged service")
    parent = SOCKET_PATH.parent.lstat()
    if not stat.S_ISDIR(parent.st_mode) or parent.st_uid != 0 or parent.st_mode & 0o022:
        raise PermissionError("watch readback socket directory is not protected")
    # systemd RuntimeDirectory owns creation/removal. Never unlink an existing
    # endpoint here: an active or stale generation requires administrator repair.
    if os.path.lexists(SOCKET_PATH): raise FileExistsError("watch readback endpoint already exists")
    group_id = int(os.environ["WATCHDOG_READBACK_GROUP_ID"])
    if group_id < 1: raise ValueError("observer service group must be non-root")
    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as listener:
        listener.bind(str(SOCKET_PATH))
        os.chown(SOCKET_PATH, 0, group_id)
        os.chmod(SOCKET_PATH, 0o660)
        listener.listen(16)
        listener.settimeout(watchdog.POLL_INTERVAL_SECONDS)
        try:
            while True:
                try: stream, _ = listener.accept()
                except socket.timeout: continue
                with stream:
                    try: handle_connection(stream)
                    except Exception:
                        # Isolate each failed RPC, including unexpected handler
                        # faults. Never return a fabricated readiness response.
                        # BaseException still lets supervisor shutdown terminate.
                        LOGGER.exception("watchdog readback RPC failed; connection closed and session blocked")
                        continue
        finally: SOCKET_PATH.unlink(missing_ok=True)


if __name__ == "__main__": serve()

Synchronous readiness and idle supervision

G001/G003 call the verify-only readiness client synchronously; the privileged worker samples enforce_watchdog and returns a nonce-bound signed acknowledgement only after its protected spool write, fsync and exact readback. Clients verify that acknowledgement before observation or action without receiving private keys or spool access. Also schedule this worker continuously under a host supervisor independent of model activity; otherwise a session that stops making calls has no demonstrated task-time or denial-limit containment. Bind the worker lifecycle, configured sample interval and telemetry collector generation to the environment's signed deployment profile, and verify liveness and terminal removal/lease revocation. A passing sample is evidence for that instant, not proof that the persistent worker is running.

Interface upgrade

The signed watchdog policy is now aidefend.desktop-watchdog-policy.v2 and requires readback_client_uids and readback_timeout_seconds. Reissue it under the existing policy authority and restart worker/client deployments together; v1 policies are rejected. The signed spool envelope remains v1; a watch sample now carries a nullable challenge_nonce (null for supervisor samples, the exact request nonce for readiness RPC). The new request schema is aidefend.desktop-watch-readback-request.v1. Verify non-root clients succeed with only public trust, unauthorized UIDs and altered/replayed challenges fail, spool failure returns no successful acknowledgement, and idle supervision remains live independently of RPC traffic.

Connection failure and scheduling limits

An authorized peer disconnecting before a complete frame closes only that RPC; the worker continues accepting clients. A malformed telemetry signature is rejected as invalid evidence and invokes the existing blocked-session containment path. This reference loop serves one request at a time. A slow peer, telemetry request, termination or audit retry can delay other clients until their signed readback deadline expires, leaving them blocked. For independently scheduled workspaces, deploy separately supervised worker instances in administrator-provisioned mount namespaces that bind the fixed socket/spool paths to distinct protected directories, and give each client the corresponding trusted socket view. Bind each instance to its protected runtime profile and signed collector identity/generation; qualify its capacity and idle supervision. The v2 watchdog policy has no task-scope field. Actual task/workspace bindings come from verified launch context, signed telemetry and nonce-bound readback, so do not claim policy-enforced per-task admission from this snippet. Do not claim concurrent readiness from this serial loop.

Reject unsupported JSON and retain per-connection failures

Both framed requests and HTTP telemetry normalize parser/canonicalization recursion or numeric-range errors into ValueError. For telemetry this follows the existing FAIL-spool and kill-switch path; malformed requests cannot name an authenticated workspace and close their RPC. The worker also logs unexpected Exception failures per connection with a traceback through Python logging, then continues accepting other clients. Configure the independently supervised service to retain these error logs and alert on repeated parser, telemetry or handler failures. A failed connection returns no readiness acknowledgement; process shutdown remains controlled by the supervisor. Verify nested request/telemetry JSON, oversized integers, unexpected handler faults, retained FAIL records and containment invocation, followed by a valid request for another workspace.

Open-source tools

• Open Policy Agent (OPA)
• OpenTelemetry
• Tesseract OCR
• pytesseract
• OpenCV
• Pillow
• Podman
• Qubes OS (disposable qube workspace isolation)
• PyGObject
• at-spi2-core
• PyNaCl (signed policy and evidence verification)
• PyYAML (policy parsing; not an isolation boundary)
• Requests (observer and evidence-service transport; not an isolation boundary)

Commercial / hosted tools

• Microsoft Intune
• Jamf Pro
• Citrix DaaS
• Omnissa Horizon

Threat mappings

MITRE ATLAS

• AML.T0051.001 LLM Prompt Injection: Indirect
• AML.T0053 AI Agent Tool Invocation
• AML.T0086 Exfiltration via AI Agent Tool Invocation
• AML.T0089 Enterprise Environment Discovery (disposable workspaces with private PID namespaces limit enumeration of host processes, sessions, and installed applications)
• AML.T0100 AI Agent Clickbait
• AML.T0101 Data Destruction via AI Agent Tool Invocation
• AML.T0112 Machine Compromise
• AML.T0112.000 Machine Compromise: Local AI Agent
• AML.T0113 Steal Web Session Cookie
• AML.T0091.001 Use Alternate Authentication Material: Web Session Cookie

MAESTRO

• Agent Tool Misuse (L7)
• Integration Risks (L7)
• Data Exfiltration (L2)
• Compromised Agents (L7)

OWASP LLM Top 10 2026

• LLM01:2026 Prompt Injection
• LLM02:2026 Sensitive Information Disclosure
• LLM03:2026 Excessive Agency
• LLM10:2026 Improper Output Handling (action-bound confirmation prevents model output from directly controlling desktop sinks)

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI01:2026 Agent Goal Hijack
• ASI02:2026 Tool Misuse and Exploitation
• ASI03:2026 Identity and Privilege Abuse
• ASI05:2026 Unexpected Code Execution (RCE)
• ASI09:2026 Human-Agent Trust Exploitation

NIST Adversarial Machine Learning 2025

• NISTAML.015 Indirect Prompt Injection
• NISTAML.036 Leaking information from user interactions
• NISTAML.039 Compromising connected resources

Cisco Integrated AI Security and Safety Framework

• AISubtech-1.3.1 Image-Text Injection (independent screenshot classification gates high-risk actions affected by image-carried instructions)
• AISubtech-12.1.3 Unsafe System / Browser / File Execution (desktop workspace and action boundaries contain coerced OS, file and network operations)
• AISubtech-14.1.1 Credential Theft (logged-out workspaces and secret, clipboard, file and egress boundaries restrict disclosure)
• AISubtech-14.1.2 Insufficient Access Controls (host isolation and authenticated privileged UI mediation restrict desktop access)
• AISubtech-2.1.1 Execution Approval Bypass (the desktop adapter binds current approval to the exact observed consequential action)
• AISubtech-2.1.2 Capability and Permission Overreach (task-bound app, file, resource and action allowlists restrict desktop capability overreach)
• AISubtech-4.2.2 Session Boundary Violation (disposable desktop state and cleanup prevent carryover between tasks)
• AISubtech-8.2.3 Data Exfiltration via Agent Tooling (logged-out workspaces and secret, clipboard, file and egress boundaries restrict disclosure)
• AISubtech-9.1.2 Unauthorized or Unsolicited System Access (desktop workspace and action boundaries contain coerced OS, file and network operations)
• AISubtech-9.1.3 Unauthorized or Unsolicited Network Access (desktop workspace and action boundaries contain coerced OS, file and network operations)
• AITech-1.2 Indirect Prompt Injection (independent screenshot classification gates high-risk actions affected by image-carried instructions)
• AITech-1.3 Multimodal Injection and Manipulation (independent screenshot classification gates high-risk actions affected by image-carried instructions)
• AITech-12.1 Tool Exploitation (desktop workspace and action boundaries contain coerced OS, file and network operations)
• AITech-14.1 Unauthorized Access (host isolation and authenticated privileged UI mediation restrict desktop access)
• AITech-2.1 Excessive Agency (the desktop adapter binds current approval to the exact observed consequential action)
• AITech-8.2 Data Exfiltration / Exposure (logged-out workspaces and secret, clipboard, file and egress boundaries restrict disclosure)
• AITech-8.3 Information Disclosure (logged-out workspaces and secret, clipboard, file and egress boundaries restrict disclosure)
• AITech-9.1 Model or Agentic System Manipulation (desktop workspace and action boundaries contain coerced OS, file and network operations)

Google Secure AI Framework 2.0 - Risks

• PIJ: Prompt Injection
• SDD: Sensitive Data Disclosure
• IIC: Insecure Integrated Component
• RA: Rogue Actions

Databricks AI Security Framework 3.0

• Agents - Core 13.2: Tool Misuse
• Agents - Core 13.3: Privilege Compromise
• Agents - Core 13.11: Unexpected RCE and Code Attacks