AID-R-007: External Side-Effect Reconciliation & Compensation

Version 1.20261015 · Actionable technique

After containment, reconcile the complete incident-scoped population of externally issued agent actions against authoritative downstream state, then idempotently cancel, reverse, compensate, or explicitly disposition each unintended, duplicate, or uncertain effect under a signed incident policy. This applies to payments, messages, IAM changes, tickets, deployments, and external data mutations.

Irreversible effects: a delivered message, disclosure, physical action, or other irreversible effect cannot be erased. Compensation or signed acceptance records its final disposition; it does not claim restoration of the prior reality.

Pillars: app, infra

Lifecycle: response, improvement

Scope boundary

Owns incident-time reconciliation of externally issued agent actions against authoritative downstream state and the idempotent cancellation, reversal, compensation, or signed final disposition of each unintended, duplicate, or uncertain effect. It cannot erase an irreversible message, disclosure, or physical action; compensation or signed acceptance records the final disposition without claiming restoration of the prior reality.

AID-H-018.008: AID-R-007 reconciles and remediates external effects after an incident; AID-H-018.008 designs stable action identity, idempotency, transaction safety, and compensability before execution. R-007 consumes the action ledger or outbox and provider identifiers produced by AID-H-018.008, but preventive transaction design does not prove that an already-issued effect was reversed.

AID-E-005: AID-R-007 remediates effects already issued to downstream systems; AID-E-005 removes compromised internal durable sessions, memory, registrations, queues, and schedules. Deleting an internal job or session does not cancel a payment, message, IAM change, deployment, or other external effect that was already accepted downstream.

AID-R-006: AID-R-007 proves the final disposition of each incident-scoped external effect; AID-R-006 orchestrates dependency-ordered recovery and cutover of a multi-component AI system. R-006 may schedule R-007 as a prerequisite, but successful system cutover does not itself prove cancellation, reversal, or compensation of prior downstream actions.

AID-R-004: AID-R-007 resolves existing incident-scoped external effects; AID-R-004 propagates the technical fix that prevents recurrence. Institutionalizing the fix does not reverse actions already issued, while compensating those actions does not prove the weakness was removed from the fleet.

Implementation guidance

AID-R-007-G001: After the containment owner freezes new dispatch, build the action population as the union of the signed action ledger, outbox or broker or workflow history, and every declared downstream provider index; keep orphaned, timed-out, and unknown actions in scope.
AID-R-007-G002: Use separate read-only provider identities to classify the authoritative state of every action and issue one policy-bound disposition decision per action.
AID-R-007-G003: Execute cancel, reverse, compensate, or approved exception handling with a stable compensation ID, then read every final downstream state through a separate provider read-only identity.

AID-R-007-G001

After the containment owner freezes new dispatch, build the action population as the union of the signed action ledger, outbox or broker or workflow history, and every declared downstream provider index; keep orphaned, timed-out, and unknown actions in scope.

Freeze dispatch and authenticate the static plan

The incident plan names only sources, target systems, state allowlists, signer roles, and policy time bounds. It deliberately does not predict hashes for readbacks, decisions, or final evidence that do not exist yet. A fixed root-owned bootstrap key authenticates the runtime profile, signer-key paths and digests, bounded Cosign/network deadlines, mTLS identity, and stage signer.

{
  "schema": "aidefend.r007.bootstrap.v1",
  "version": "2026-07-20.1",
  "maximum_signed_document_bytes": 8388608,
  "cosign_timeout_seconds": 30,
  "maximum_readback_staleness_seconds": 300,
  "maximum_future_skew_seconds": 30,
  "stage_signing_key_uri": "awskms:///alias/aidefend-r007-stage",
  "executor_origin": "https://reconciliation-executor.internal.example",
  "connect_timeout_seconds": 3,
  "read_timeout_seconds": 20,
  "maximum_response_bytes": 65536,
  "client_cert": "/etc/aidefend/r007/executor-client.crt",
  "client_key": "/etc/aidefend/r007/executor-client.key",
  "ca_file": "/etc/aidefend/r007/internal-ca.pem",
  "trust": {
    "incident_policy": {"path":"/etc/aidefend/r007/trust/incident-policy.pub","sha256":"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"},
    "r007_stage": {"path":"/etc/aidefend/r007/trust/r007-stage.pub","sha256":"123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef0"},
    "ledger_exporter": {"path":"/etc/aidefend/r007/trust/ledger-exporter.pub","sha256":"23456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef01"},
    "provider_reader": {"path":"/etc/aidefend/r007/trust/provider-reader.pub","sha256":"3456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef012"},
    "decision_authority": {"path":"/etc/aidefend/r007/trust/decision-authority.pub","sha256":"456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123"}
  }
}

{
  "schema_version": "aidefend.external-effect-reconciliation.v2",
  "source_policy_version": "INC-2041-r3",
  "incident_id": "INC-2041",
  "tenant_id": "tenant-acme",
  "required_sources": {
    "action_ledger": {"path":"/srv/incidents/INC-2041/action-ledger.jsonl","bundle":"/srv/incidents/INC-2041/action-ledger.jsonl.sigstore.json","signer_id":"ledger_exporter","signer_key_sha256":"23456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef01","sha256":"56789abcdef0123456789abcdef0123456789abcdef0123456789abcdef01234"}
  },
  "target_systems": ["payment-provider"],
  "authoritative_state_allowlist": {"payment-provider":["ABSENT","AUTHORIZED","CAPTURED","REVERSED"]},
  "initial_readback_signer_id": "provider_reader",
  "decision_signer_id": "decision_authority",
  "final_readback_signer_id": "provider_reader",
  "max_readback_staleness_seconds": 300,
  "max_future_skew_seconds": 30
}

Verify signed source bytes and build the exact union

# File: recovery/r007_common.py
from __future__ import annotations

import hashlib
import json
import math
import os
import re
import stat
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.parse import urlsplit

BOOTSTRAP_PATH = Path("/etc/aidefend/r007/bootstrap.json")
BOOTSTRAP_BUNDLE = Path(str(BOOTSTRAP_PATH) + ".sigstore.json")
BOOTSTRAP_ROOT = Path("/etc/aidefend/r007/bootstrap-root.pub")
# Compiled process-safety ceilings apply before the signed bootstrap can supply narrower bounds.
BOOTSTRAP_CAPTURE_CEILING = 1024 * 1024
BOOTSTRAP_VERIFICATION_DEADLINE_CEILING_SECONDS = 30.0
SIGNED_INPUT_CAPTURE_CEILING = 16 * 1024 * 1024
COSIGN_DEADLINE_CEILING_SECONDS = 120.0
NETWORK_DEADLINE_CEILING_SECONDS = 60.0
SHA = re.compile(r"^[0-9a-f]{64}$")
SAFE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,254}$")
IDENTITY = ("incident_id","tenant_id","action_id","target_system","target_resource_id","idempotency_key")
DESCRIPTOR_FIELDS = {"path","bundle","signer_id","signer_key_sha256","sha256"}
PLAN_FIELDS = {"schema_version","source_policy_version","incident_id","tenant_id","required_sources","target_systems","authoritative_state_allowlist","initial_readback_signer_id","decision_signer_id","final_readback_signer_id","max_readback_staleness_seconds","max_future_skew_seconds"}
POPULATION_FIELDS = {"schema_version","plan_sha256","incident_id","tenant_id","source_sha256","population_sha256","population"}
POPULATION_ROW_FIELDS = set(IDENTITY) | {"observed_sources"}
BOOTSTRAP_FIELDS = {"schema","version","maximum_signed_document_bytes","cosign_timeout_seconds","maximum_readback_staleness_seconds","maximum_future_skew_seconds","stage_signing_key_uri","executor_origin","connect_timeout_seconds","read_timeout_seconds","maximum_response_bytes","client_cert","client_key","ca_file","trust"}
TRUST_FIELDS = {"path","sha256"}

def canonical(value: object) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",",":"), ensure_ascii=False, allow_nan=False).encode("utf-8")

def digest(value: object, label: str) -> str:
    if not isinstance(value,str) or SHA.fullmatch(value) is None or set(value) == {"0"}: raise ValueError(f"{label} must be a nonzero SHA-256")
    return value

def strict_json(raw: bytes, label: str) -> Any:
    def pairs(items: list[tuple[str,Any]]) -> dict[str,Any]:
        value: dict[str,Any] = {}
        for key,item in items:
            if key in value: raise ValueError(f"{label}: duplicate key {key}")
            value[key] = item
        return value
    def constant(value: str) -> None: raise ValueError(f"{label}: non-finite number {value}")
    return json.loads(raw.decode("utf-8","strict"), object_pairs_hook=pairs, parse_constant=constant)

def strict_object(raw: bytes, label: str) -> dict[str,Any]:
    value = strict_json(raw,label)
    if not isinstance(value,dict): raise ValueError(f"{label} must contain one object")
    return value

def bounded_number(value: object, label: str, ceiling: float, *, zero: bool = False) -> float:
    if isinstance(value,bool) or not isinstance(value,(int,float)) or not math.isfinite(float(value)) or value < 0 or (value == 0 and not zero) or value > ceiling: raise ValueError(f"{label} is invalid")
    return float(value)

def fixed_file(path: Path, label: str) -> bytes:
    info = path.lstat()
    if path.is_symlink() or not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o022: raise RuntimeError(f"{label} must be a fixed root-owned regular file")
    return path.read_bytes()

def read_limited(path: Path, maximum: int) -> bytes:
    if path.is_symlink() or not path.is_file(): raise ValueError(f"{path} must be a non-symlink regular file")
    with path.open("rb") as handle: raw = handle.read(maximum + 1)
    if not raw or len(raw) > maximum: raise ValueError(f"{path} is empty or exceeds its authenticated bound")
    return raw

def private_file(path: Path, raw: bytes) -> None:
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os,"O_NOFOLLOW",0)
    fd = os.open(path,flags,0o400)
    with os.fdopen(fd,"wb") as handle: handle.write(raw); handle.flush(); os.fsync(handle.fileno())

def verify_raw(payload_raw: bytes, bundle_raw: bytes, key_raw: bytes, timeout: float, label: str) -> None:
    with tempfile.TemporaryDirectory(prefix="r007-verify-") as directory:
        root = Path(directory); os.chmod(root,0o700)
        payload,bundle,key = root/"payload",root/"payload.sigstore.json",root/"key.pub"
        private_file(payload,payload_raw); private_file(bundle,bundle_raw); private_file(key,key_raw)
        subprocess.run(["cosign","verify-blob","--key",str(key),"--bundle",str(bundle),str(payload)], check=True, capture_output=True, text=True, timeout=timeout)
        if payload.read_bytes()!=payload_raw or bundle.read_bytes()!=bundle_raw or key.read_bytes()!=key_raw: raise RuntimeError(f"{label} private verified snapshot changed")

def load_bootstrap() -> tuple[dict[str,Any],str]:
    key_raw = fixed_file(BOOTSTRAP_ROOT,"bootstrap root")
    raw,bundle = read_limited(BOOTSTRAP_PATH,BOOTSTRAP_CAPTURE_CEILING),read_limited(BOOTSTRAP_BUNDLE,BOOTSTRAP_CAPTURE_CEILING)
    verify_raw(raw,bundle,key_raw,BOOTSTRAP_VERIFICATION_DEADLINE_CEILING_SECONDS,"bootstrap")
    value = strict_object(raw,"bootstrap")
    if set(value)!=BOOTSTRAP_FIELDS or value.get("schema")!="aidefend.r007.bootstrap.v1" or not isinstance(value["version"],str) or not value["version"]: raise ValueError("bootstrap schema differs")
    maximum = value["maximum_signed_document_bytes"]
    if isinstance(maximum,bool) or not isinstance(maximum,int) or not 1 <= maximum <= SIGNED_INPUT_CAPTURE_CEILING: raise ValueError("bootstrap byte bound differs")
    bounded_number(value["cosign_timeout_seconds"],"cosign timeout",COSIGN_DEADLINE_CEILING_SECONDS)
    bounded_number(value["maximum_readback_staleness_seconds"],"readback ceiling",86400)
    bounded_number(value["maximum_future_skew_seconds"],"future-skew ceiling",3600,zero=True)
    bounded_number(value["connect_timeout_seconds"],"connect timeout",NETWORK_DEADLINE_CEILING_SECONDS)
    bounded_number(value["read_timeout_seconds"],"read timeout",NETWORK_DEADLINE_CEILING_SECONDS)
    response_max=value["maximum_response_bytes"]
    if isinstance(response_max,bool) or not isinstance(response_max,int) or not 1 <= response_max <= 1024*1024: raise ValueError("response byte bound differs")
    endpoint=urlsplit(value["executor_origin"])
    if endpoint.scheme!="https" or endpoint.hostname is None or endpoint.port not in (None,443) or endpoint.username or endpoint.password or endpoint.path not in ("", "/") or endpoint.query or endpoint.fragment: raise ValueError("executor origin is not canonical HTTPS")
    for field in ("client_cert","client_key","ca_file"): fixed_file(Path(value[field]),field)
    trust=value["trust"]
    if not isinstance(trust,dict) or not {"incident_policy","r007_stage"} <= set(trust) or len(trust)>64: raise ValueError("bootstrap trust population differs")
    for signer,entry in trust.items():
        if SAFE.fullmatch(signer) is None or not isinstance(entry,dict) or set(entry)!=TRUST_FIELDS: raise ValueError("bootstrap trust entry differs")
        key=fixed_file(Path(entry["path"]),f"{signer} key")
        if hashlib.sha256(key).hexdigest()!=digest(entry["sha256"],f"{signer} key digest"): raise RuntimeError(f"{signer} key digest differs")
    if not isinstance(value["stage_signing_key_uri"],str) or not value["stage_signing_key_uri"]: raise ValueError("stage signing key URI is absent")
    return value,hashlib.sha256(raw).hexdigest()

def verify_pair(path: Path, bundle: Path, signer_id: str, bootstrap: dict[str,Any], label: str) -> bytes:
    if bundle != Path(str(path)+".sigstore.json"): raise ValueError(f"{label} bundle must be sibling <payload>.sigstore.json")
    entry=bootstrap["trust"].get(signer_id)
    if not isinstance(entry,dict): raise ValueError(f"{label} signer is not trusted")
    maximum=bootstrap["maximum_signed_document_bytes"]
    raw,bundle_raw=read_limited(path,maximum),read_limited(bundle,maximum)
    key_raw=fixed_file(Path(entry["path"]),f"{signer_id} key")
    if hashlib.sha256(key_raw).hexdigest()!=entry["sha256"]: raise RuntimeError(f"{label} signer key changed")
    verify_raw(raw,bundle_raw,key_raw,bootstrap["cosign_timeout_seconds"],label)
    return raw

def descriptor(value: object, bootstrap: dict[str,Any], label: str) -> dict[str,Any]:
    if not isinstance(value,dict) or set(value)!=DESCRIPTOR_FIELDS: raise ValueError(f"{label} descriptor schema differs")
    signer=value["signer_id"]
    if signer not in bootstrap["trust"] or value["signer_key_sha256"]!=bootstrap["trust"][signer]["sha256"]: raise ValueError(f"{label} signer trust binding differs")
    path,bundle=Path(value["path"]),Path(value["bundle"])
    if not path.is_absolute() or bundle!=Path(str(path)+".sigstore.json"): raise ValueError(f"{label} paths must be absolute sibling paths")
    digest(value["sha256"],f"{label} payload")
    return value

def verify_descriptor(value: object, bootstrap: dict[str,Any], label: str) -> bytes:
    desc=descriptor(value,bootstrap,label)
    raw=verify_pair(Path(desc["path"]),Path(desc["bundle"]),desc["signer_id"],bootstrap,label)
    if hashlib.sha256(raw).hexdigest()!=desc["sha256"]: raise RuntimeError(f"{label} payload digest differs")
    return raw

def text(value: object, label: str) -> str:
    if not isinstance(value,str) or not value or len(value)>512 or any(ord(ch)<32 for ch in value): raise ValueError(f"{label} is invalid")
    return value

def identity(row: dict[str,Any], label: str) -> tuple[str,...]:
    return tuple(text(row.get(field),f"{label}.{field}") for field in IDENTITY)

def parse_plan(raw: bytes, bootstrap: dict[str,Any]) -> dict[str,Any]:
    plan=strict_object(raw,"incident plan")
    if set(plan)!=PLAN_FIELDS or plan.get("schema_version")!="aidefend.external-effect-reconciliation.v2" or not isinstance(plan["source_policy_version"],str) or not plan["source_policy_version"]: raise ValueError("incident plan schema differs")
    text(plan["incident_id"],"incident_id"); text(plan["tenant_id"],"tenant_id")
    required=plan["required_sources"]
    if not isinstance(required,dict) or not required or list(required)!=sorted(required) or any(SAFE.fullmatch(name) is None for name in required): raise ValueError("required source population is not sorted and exact")
    for name,value in required.items(): descriptor(value,bootstrap,f"required_sources.{name}")
    targets=plan["target_systems"]
    if not isinstance(targets,list) or not targets or targets!=sorted(set(targets)) or any(not isinstance(item,str) or not item for item in targets): raise ValueError("target system population differs")
    allow=plan["authoritative_state_allowlist"]
    if not isinstance(allow,dict) or set(allow)!=set(targets) or any(not isinstance(states,list) or not states or states!=sorted(set(states)) for states in allow.values()): raise ValueError("authoritative-state allowlist differs")
    for role in ("initial_readback_signer_id","decision_signer_id","final_readback_signer_id"):
        if plan[role] not in bootstrap["trust"]: raise ValueError(f"{role} is not trusted")
    stale=bounded_number(plan["max_readback_staleness_seconds"],"plan readback age",bootstrap["maximum_readback_staleness_seconds"])
    future=bounded_number(plan["max_future_skew_seconds"],"plan future skew",bootstrap["maximum_future_skew_seconds"],zero=True)
    plan["max_readback_staleness_seconds"],plan["max_future_skew_seconds"]=stale,future
    return plan

def plan_input(path: Path,bundle: Path,bootstrap: dict[str,Any]) -> tuple[dict[str,Any],bytes]:
    raw=verify_pair(path,bundle,"incident_policy",bootstrap,"incident plan")
    return parse_plan(raw,bootstrap),raw

def jsonl(raw: bytes,label: str) -> list[dict[str,Any]]:
    rows=[]
    for number,line in enumerate(raw.splitlines(),1):
        if not line: raise ValueError(f"{label}:{number} blank lines are forbidden")
        value=strict_json(line,f"{label}:{number}")
        if not isinstance(value,dict): raise ValueError(f"{label}:{number} must be an object")
        rows.append(value)
    return rows

def parse_population(raw: bytes, plan: dict[str,Any], plan_sha: str) -> tuple[dict[str,Any],dict[str,tuple[str,...]]]:
    value=strict_object(raw,"action population")
    if set(value)!=POPULATION_FIELDS or value.get("schema_version")!="aidefend.action-population.v2" or value["plan_sha256"]!=plan_sha or (value["incident_id"],value["tenant_id"])!=(plan["incident_id"],plan["tenant_id"]): raise ValueError("action population envelope differs")
    if value["source_sha256"]!={name:plan["required_sources"][name]["sha256"] for name in sorted(plan["required_sources"])}: raise ValueError("action population source digests differ")
    rows=value["population"]
    if not isinstance(rows,list) or not rows or hashlib.sha256(canonical(rows)).hexdigest()!=value["population_sha256"]: raise ValueError("action population digest differs")
    by_id={}
    for row in rows:
        if not isinstance(row,dict) or set(row)!=POPULATION_ROW_FIELDS: raise ValueError("action population row schema differs")
        bound=identity(row,"population"); action=bound[2]
        if action in by_id or bound[0]!=plan["incident_id"] or bound[1]!=plan["tenant_id"] or bound[3] not in plan["target_systems"]: raise ValueError("action population identity differs")
        sources=row["observed_sources"]
        if not isinstance(sources,list) or not sources or sources!=sorted(set(sources)) or not set(sources)<=set(plan["required_sources"]): raise ValueError("observed source membership differs")
        by_id[action]=bound
    if [row["action_id"] for row in rows]!=sorted(by_id): raise ValueError("action population is not sorted")
    return value,by_id

def utc(value: object,label: str) -> datetime:
    if not isinstance(value,str): raise ValueError(f"{label} must be RFC 3339")
    try: parsed=datetime.fromisoformat(value.replace("Z","+00:00"))
    except ValueError as error: raise ValueError(f"{label} must be RFC 3339") from error
    if parsed.tzinfo is None: raise ValueError(f"{label} lacks timezone")
    return parsed.astimezone(timezone.utc)

def fresh(value: object,label: str,now: datetime,plan: dict[str,Any]) -> str:
    parsed=utc(value,label); age=(now-parsed).total_seconds()
    if age>plan["max_readback_staleness_seconds"] or age < -plan["max_future_skew_seconds"]: raise ValueError(f"{label} is stale or future-dated")
    return parsed.isoformat().replace("+00:00","Z")

def publish(output: Path,value: dict[str,Any],bootstrap: dict[str,Any]) -> str:
    bundle=Path(str(output)+".sigstore.json")
    if output.exists() or bundle.exists(): raise FileExistsError("create-once output or bundle already exists")
    parent=output.parent
    info=parent.lstat()
    if parent.is_symlink() or not stat.S_ISDIR(info.st_mode) or info.st_uid!=os.geteuid() or info.st_mode&0o022: raise RuntimeError("output directory is not private and service-owned")
    raw=canonical(value)+b"\n"
    with tempfile.TemporaryDirectory(prefix=".r007-publish-",dir=parent) as directory:
        root=Path(directory);os.chmod(root,0o700);payload=root/"payload";bundle_copy=root/"payload.sigstore.json"
        private_file(payload,raw)
        subprocess.run(["cosign","sign-blob","--yes","--key",bootstrap["stage_signing_key_uri"],"--bundle",str(bundle_copy),str(payload)],check=True,capture_output=True,text=True,timeout=bootstrap["cosign_timeout_seconds"])
        verify_pair(payload,bundle_copy,"r007_stage",bootstrap,"stage output")
        if payload.read_bytes()!=raw: raise RuntimeError("signed stage output changed")
        os.link(bundle_copy,bundle,follow_symlinks=False)
        os.link(payload,output,follow_symlinks=False)  # payload is the create-once commit marker
    return hashlib.sha256(raw).hexdigest()

# File: recovery/build_action_population.py
from __future__ import annotations
import argparse,hashlib,json
from collections import defaultdict
from pathlib import Path
from recovery.r007_common import IDENTITY,canonical,identity,jsonl,load_bootstrap,plan_input,publish,verify_descriptor

def main() -> None:
    parser=argparse.ArgumentParser();parser.add_argument("--plan",type=Path,required=True);parser.add_argument("--plan-bundle",type=Path,required=True);parser.add_argument("--output",type=Path,required=True);args=parser.parse_args()
    bootstrap,bootstrap_sha=load_bootstrap();plan,plan_raw=plan_input(args.plan,args.plan_bundle,bootstrap)
    union:dict[str,tuple[str,...]]={};membership:dict[str,set[str]]=defaultdict(set);source_sha={}
    for source,desc in plan["required_sources"].items():
        raw=verify_descriptor(desc,bootstrap,source);source_sha[source]=hashlib.sha256(raw).hexdigest()
        seen=set()
        for row in jsonl(raw,source):
            bound=identity(row,source);action=bound[2]
            if action in seen: raise ValueError(f"{source}: duplicate action_id {action}")
            seen.add(action)
            if bound[0]!=plan["incident_id"] or bound[1]!=plan["tenant_id"] or bound[3] not in plan["target_systems"]: raise ValueError(f"{source}:{action} is outside the incident plan")
            if action in union and union[action]!=bound: raise ValueError(f"{action}: cross-source identity conflict")
            union[action]=bound;membership[action].add(source)
    if not union: raise RuntimeError("declared sources produced an empty action population")
    rows=[]
    for action in sorted(union):
        row=dict(zip(IDENTITY,union[action]));row["observed_sources"]=sorted(membership[action]);rows.append(row)
    value={"schema_version":"aidefend.action-population.v2","plan_sha256":hashlib.sha256(plan_raw).hexdigest(),"incident_id":plan["incident_id"],"tenant_id":plan["tenant_id"],"source_sha256":source_sha,"population_sha256":hashlib.sha256(canonical(rows)).hexdigest(),"population":rows}
    print(json.dumps({"outcome":"PASS","population_sha256":value["population_sha256"],"receipt_sha256":publish(args.output,value,bootstrap),"bootstrap_sha256":bootstrap_sha},sort_keys=True))

if __name__=="__main__": main()

set -euo pipefail
umask 077
python recovery/build_action_population.py \
  --plan policy/reconciliation-plan.json \
  --plan-bundle policy/reconciliation-plan.json.sigstore.json \
  --output evidence/action-population.json

The population output is signed, verified from private snapshots, and published bundle-first with the payload as a create-once commit marker. Duplicate IDs, cross-source identity conflicts, provider-only orphans, ledger-only timeouts, changed bytes, unknown signer keys, source omission, signature timeout, or an empty population prevents PASS.

AID-R-007-G002

Use separate read-only provider identities to classify the authoritative state of every action and issue one policy-bound disposition decision per action.

Classify every action after the population exists

After G001 commits the population, separate read-only provider identities issue an authoritative-readback artifact and the decision authority issues a validity-bounded decision artifact. Neither artifact is pre-bound to a future hash in the incident plan; each is verified when it exists against a fixed signer role and key digest.

# File: recovery/classify_actions.py
from __future__ import annotations
import argparse,hashlib,json
from datetime import datetime,timezone
from pathlib import Path
from recovery.r007_common import IDENTITY,canonical,digest,fresh,identity,load_bootstrap,parse_population,plan_input,publish,strict_object,text,utc,verify_pair

INITIAL_FIELDS={"schema_version","incident_id","tenant_id","population_sha256","rows"}; INITIAL_ROW=set(IDENTITY)|{"authoritative_state","state_sha256","provider_receipt_id","observed_at"}
DECISION_FIELDS={"schema_version","incident_id","tenant_id","population_sha256","issued_at","expires_at","rows"}; DECISION_ROW=set(IDENTITY)|{"authoritative_state_sha256","final_disposition","expected_state_sha256","compensation_id","exception_approval_id"}
FINAL={"INTENDED_CONFIRMED","ABSENT_CONFIRMED","CANCELLED","REVERSED","COMPENSATED","ACCEPTED_EXCEPTION"}; REMEDIATING={"CANCELLED","REVERSED","COMPENSATED"}

def indexed(value:dict,fields:set,row_fields:set,schema:str,population:dict,bindings:dict,label:str)->dict:
    if set(value)!=fields or value.get("schema_version")!=schema or (value["incident_id"],value["tenant_id"])!=(population["incident_id"],population["tenant_id"]) or value["population_sha256"]!=population["population_sha256"]: raise ValueError(f"{label} envelope differs")
    rows=value["rows"]
    if not isinstance(rows,list) or len(rows)!=len(bindings): raise ValueError(f"{label} population count differs")
    result={}
    for row in rows:
        if not isinstance(row,dict) or set(row)!=row_fields: raise ValueError(f"{label} row schema differs")
        bound=identity(row,label);action=bound[2]
        if action in result or bindings.get(action)!=bound: raise ValueError(f"{label} identity/population differs")
        result[action]=row
    if set(result)!=set(bindings): raise ValueError(f"{label} complete membership differs")
    return result

def main()->None:
    p=argparse.ArgumentParser()
    for name in ("plan","plan-bundle","population","population-bundle","initial-readback","initial-readback-bundle","decisions","decisions-bundle"): p.add_argument("--"+name,type=Path,required=True)
    p.add_argument("--output",type=Path,required=True);a=p.parse_args();bootstrap,bootstrap_sha=load_bootstrap();plan,plan_raw=plan_input(a.plan,a.plan_bundle,bootstrap);plan_sha=hashlib.sha256(plan_raw).hexdigest()
    pop_raw=verify_pair(a.population,a.population_bundle,"r007_stage",bootstrap,"action population");population,bindings=parse_population(pop_raw,plan,plan_sha)
    initial_raw=verify_pair(a.initial_readback,a.initial_readback_bundle,plan["initial_readback_signer_id"],bootstrap,"initial readback")
    decision_raw=verify_pair(a.decisions,a.decisions_bundle,plan["decision_signer_id"],bootstrap,"decisions")
    initials=indexed(strict_object(initial_raw,"initial readback"),INITIAL_FIELDS,INITIAL_ROW,"aidefend.authoritative-readback.v2",population,bindings,"initial readback")
    decision_doc=strict_object(decision_raw,"decisions");decisions=indexed(decision_doc,DECISION_FIELDS,DECISION_ROW,"aidefend.reconciliation-decisions.v2",population,bindings,"decisions")
    now=datetime.now(timezone.utc);issued=utc(decision_doc["issued_at"],"decisions issued_at");expires=utc(decision_doc["expires_at"],"decisions expires_at")
    if (issued-now).total_seconds()>plan["max_future_skew_seconds"] or expires<=now or expires<=issued: raise ValueError("decision artifact is future-issued, expired, or inverted")
    compensation=set();rows=[]
    for action in sorted(bindings):
        initial,decision=initials[action],decisions[action];state=text(initial["authoritative_state"],f"{action}.state");state_sha=digest(initial["state_sha256"],f"{action}.state_sha256")
        if state not in plan["authoritative_state_allowlist"][bindings[action][3]] or decision["authoritative_state_sha256"]!=state_sha: raise ValueError(f"{action}: authoritative state classification differs")
        text(initial["provider_receipt_id"],f"{action}.provider_receipt_id");fresh(initial["observed_at"],f"{action}.observed_at",now,plan)
        disposition=decision["final_disposition"];digest(decision["expected_state_sha256"],f"{action}.expected_state_sha256")
        if disposition not in FINAL: raise ValueError(f"{action}: unsupported disposition")
        comp,approval=decision["compensation_id"],decision["exception_approval_id"]
        if disposition in REMEDIATING:
            comp=text(comp,f"{action}.compensation_id")
            if comp in compensation or approval not in (None,""): raise ValueError(f"{action}: compensation is reused or mixed with exception")
            compensation.add(comp)
        elif disposition=="ACCEPTED_EXCEPTION":
            text(approval,f"{action}.exception_approval_id")
            if comp not in (None,""): raise ValueError(f"{action}: exception has compensation ID")
        elif comp not in (None,"") or approval not in (None,""): raise ValueError(f"{action}: unexpected remediation metadata")
        rows.append({"action_id":action,"authoritative_state":state,"authoritative_state_sha256":state_sha,"final_disposition":disposition,"expected_state_sha256":decision["expected_state_sha256"],"compensation_id":comp,"exception_approval_id":approval})
    value={"schema_version":"aidefend.reconciliation-assessment.v2","outcome":"PASS","incident_id":plan["incident_id"],"tenant_id":plan["tenant_id"],"plan_sha256":plan_sha,"population_receipt_sha256":hashlib.sha256(pop_raw).hexdigest(),"population_sha256":population["population_sha256"],"initial_readback_sha256":hashlib.sha256(initial_raw).hexdigest(),"decision_sha256":hashlib.sha256(decision_raw).hexdigest(),"evaluated_at":now.isoformat().replace("+00:00","Z"),"rows":rows}
    print(json.dumps({"outcome":"PASS","assessment_sha256":publish(a.output,value,bootstrap),"bootstrap_sha256":bootstrap_sha},sort_keys=True))

if __name__=="__main__":main()

set -euo pipefail
umask 077
python recovery/classify_actions.py \
  --plan policy/reconciliation-plan.json --plan-bundle policy/reconciliation-plan.json.sigstore.json \
  --population evidence/action-population.json --population-bundle evidence/action-population.json.sigstore.json \
  --initial-readback evidence/authoritative-readback.json --initial-readback-bundle evidence/authoritative-readback.json.sigstore.json \
  --decisions evidence/reconciliation-decisions.json --decisions-bundle evidence/reconciliation-decisions.json.sigstore.json \
  --output evidence/reconciliation-assessment.json

The assessment requires exact membership and identity equality across the population, initial readbacks, and decisions; policy-allowed authoritative states; fresh provider observations; unexpired decisions; unique compensation IDs; and signed exception approvals. Missing, stale, duplicate, conflicting, pending, or extra actions keep the stage incomplete.

AID-R-007-G003

Execute cancel, reverse, compensate, or approved exception handling with a stable compensation ID, then read every final downstream state through a separate provider read-only identity.

Dispatch the signed decisions, then verify separate final readback

The dispatcher can execute only the already signed decision population through the authenticated internal executor, using the original idempotency key and stable compensation ID. It cannot manufacture final state. After dispatch, a separate read-only provider collector signs the complete final-readback population; the verifier replays every prior binding and publishes PASS only after exact final state is proven.

# File: recovery/reconcile_final.py
from __future__ import annotations
import argparse,hashlib,json
from datetime import datetime,timezone
from pathlib import Path
from typing import Any
import requests
from recovery.r007_common import IDENTITY,canonical,digest,fresh,identity,load_bootstrap,parse_population,plan_input,publish,strict_object,text,utc,verify_pair

ASSESSMENT_FIELDS={"schema_version","outcome","incident_id","tenant_id","plan_sha256","population_receipt_sha256","population_sha256","initial_readback_sha256","decision_sha256","evaluated_at","rows"}
ASSESSMENT_ROW={"action_id","authoritative_state","authoritative_state_sha256","final_disposition","expected_state_sha256","compensation_id","exception_approval_id"}
DECISION_FIELDS={"schema_version","incident_id","tenant_id","population_sha256","issued_at","expires_at","rows"}; DECISION_ROW=set(IDENTITY)|{"authoritative_state_sha256","final_disposition","expected_state_sha256","compensation_id","exception_approval_id"}
FINAL_FIELDS={"schema_version","incident_id","tenant_id","population_sha256","rows"}; FINAL_ROW=set(IDENTITY)|{"final_disposition","state_sha256","provider_receipt_id","observed_at","compensation_id","exception_approval_id"}
DISPATCH_FIELDS={"schema_version","incident_id","tenant_id","plan_sha256","population_sha256","assessment_sha256","decision_sha256","rows"}
REMEDIATING={"CANCELLED","REVERSED","COMPENSATED"}

def index_rows(doc:dict,fields:set,row_fields:set,schema:str,population:dict,bindings:dict,label:str)->dict:
    if set(doc)!=fields or doc.get("schema_version")!=schema or (doc["incident_id"],doc["tenant_id"])!=(population["incident_id"],population["tenant_id"]) or doc["population_sha256"]!=population["population_sha256"]: raise ValueError(f"{label} envelope differs")
    rows=doc["rows"]
    if not isinstance(rows,list) or len(rows)!=len(bindings): raise ValueError(f"{label} count differs")
    result={}
    for row in rows:
        if not isinstance(row,dict) or set(row)!=row_fields: raise ValueError(f"{label} row schema differs")
        bound=identity(row,label);action=bound[2]
        if action in result or bindings.get(action)!=bound: raise ValueError(f"{label} identity differs")
        result[action]=row
    if set(result)!=set(bindings): raise ValueError(f"{label} membership differs")
    return result

def bounded_json(response:requests.Response,maximum:int)->dict[str,Any]:
    if response.is_redirect or response.headers.get("Location") or response.headers.get("Content-Encoding","identity").lower()!="identity": raise RuntimeError("executor redirect/compression is forbidden")
    if response.headers.get("Content-Type","").split(";",1)[0].strip().lower()!="application/json": raise RuntimeError("executor response type differs")
    response.raise_for_status();parts=[];total=0
    for part in response.iter_content(65536):
        total+=len(part)
        if total>maximum: raise RuntimeError("executor response exceeds authenticated bound")
        parts.append(part)
    value=strict_object(b"".join(parts),"executor response")
    if set(value)!={"action_id","idempotency_key","operation","accepted","compensation_id"} or value["accepted"] is not True: raise RuntimeError("executor acknowledgement differs")
    return value

def inputs(a:argparse.Namespace):
    bootstrap,bootstrap_sha=load_bootstrap();plan,plan_raw=plan_input(a.plan,a.plan_bundle,bootstrap);plan_sha=hashlib.sha256(plan_raw).hexdigest()
    pop_raw=verify_pair(a.population,a.population_bundle,"r007_stage",bootstrap,"action population");population,bindings=parse_population(pop_raw,plan,plan_sha)
    assessment_raw=verify_pair(a.assessment,a.assessment_bundle,"r007_stage",bootstrap,"assessment");assessment=strict_object(assessment_raw,"assessment")
    decision_raw=verify_pair(a.decisions,a.decisions_bundle,plan["decision_signer_id"],bootstrap,"decisions");decision_doc=strict_object(decision_raw,"decisions");decisions=index_rows(decision_doc,DECISION_FIELDS,DECISION_ROW,"aidefend.reconciliation-decisions.v2",population,bindings,"decisions")
    if set(assessment)!=ASSESSMENT_FIELDS or assessment.get("schema_version")!="aidefend.reconciliation-assessment.v2" or assessment["outcome"]!="PASS" or assessment["plan_sha256"]!=plan_sha or assessment["population_receipt_sha256"]!=hashlib.sha256(pop_raw).hexdigest() or assessment["population_sha256"]!=population["population_sha256"] or assessment["decision_sha256"]!=hashlib.sha256(decision_raw).hexdigest(): raise ValueError("assessment input binding differs")
    now=datetime.now(timezone.utc);issued=utc(decision_doc["issued_at"],"decisions issued_at");expires=utc(decision_doc["expires_at"],"decisions expires_at")
    if (issued-now).total_seconds()>plan["max_future_skew_seconds"] or expires<=now or expires<=issued: raise ValueError("decision artifact is future-issued, expired, or inverted")
    fresh(assessment["evaluated_at"],"assessment evaluated_at",now,plan)
    assessment_rows=assessment["rows"]
    if not isinstance(assessment_rows,list) or len(assessment_rows)!=len(bindings): raise ValueError("assessment row population differs")
    observed=set()
    for row in assessment_rows:
        if not isinstance(row,dict) or set(row)!=ASSESSMENT_ROW: raise ValueError("assessment row schema differs")
        action=row["action_id"]
        if action in observed or action not in decisions: raise ValueError("assessment action membership differs")
        observed.add(action);decision=decisions[action]
        for field in ("authoritative_state_sha256","final_disposition","expected_state_sha256","compensation_id","exception_approval_id"):
            if row[field]!=decision[field]: raise ValueError(f"{action}: assessment/decision binding differs for {field}")
    if observed!=set(bindings): raise ValueError("assessment complete membership differs")
    return bootstrap,bootstrap_sha,plan,plan_sha,population,bindings,assessment_raw,decisions,decision_raw

def dispatch(a:argparse.Namespace)->None:
    bootstrap,bootstrap_sha,plan,plan_sha,population,bindings,assessment_raw,decisions,decision_raw=inputs(a);rows=[]
    with requests.Session() as session:
        session.trust_env=False
        for action in sorted(bindings):
            decision=decisions[action]
            request={"incident_id":plan["incident_id"],"tenant_id":plan["tenant_id"],"action_id":action,"target_system":bindings[action][3],"target_resource_id":bindings[action][4],"idempotency_key":bindings[action][5],"final_disposition":decision["final_disposition"],"expected_state_sha256":decision["expected_state_sha256"],"compensation_id":decision["compensation_id"],"exception_approval_id":decision["exception_approval_id"],"decision_sha256":hashlib.sha256(decision_raw).hexdigest()}
            with session.post(bootstrap["executor_origin"]+"/v1/reconciliation-actions",headers={"Accept":"application/json","Accept-Encoding":"identity","Idempotency-Key":bindings[action][5]},json=request,cert=(bootstrap["client_cert"],bootstrap["client_key"]),verify=bootstrap["ca_file"],timeout=(bootstrap["connect_timeout_seconds"],bootstrap["read_timeout_seconds"]),allow_redirects=False,stream=True) as response: observed=bounded_json(response,bootstrap["maximum_response_bytes"])
            if observed["action_id"]!=action or observed["idempotency_key"]!=bindings[action][5] or observed["operation"]!=decision["final_disposition"] or observed["compensation_id"]!=decision["compensation_id"]: raise RuntimeError(f"{action}: executor binding differs")
            rows.append(observed)
    value={"schema_version":"aidefend.reconciliation-dispatch.v2","incident_id":plan["incident_id"],"tenant_id":plan["tenant_id"],"plan_sha256":plan_sha,"population_sha256":population["population_sha256"],"assessment_sha256":hashlib.sha256(assessment_raw).hexdigest(),"decision_sha256":hashlib.sha256(decision_raw).hexdigest(),"rows":rows}
    print(json.dumps({"outcome":"PASS","dispatch_sha256":publish(a.output,value,bootstrap),"bootstrap_sha256":bootstrap_sha},sort_keys=True))

def verify(a:argparse.Namespace)->None:
    bootstrap,bootstrap_sha,plan,plan_sha,population,bindings,assessment_raw,decisions,decision_raw=inputs(a)
    dispatch_raw=verify_pair(a.dispatch_receipt,a.dispatch_receipt_bundle,"r007_stage",bootstrap,"dispatch receipt");dispatch_doc=strict_object(dispatch_raw,"dispatch receipt")
    if set(dispatch_doc)!=DISPATCH_FIELDS or dispatch_doc.get("schema_version")!="aidefend.reconciliation-dispatch.v2" or dispatch_doc["plan_sha256"]!=plan_sha or dispatch_doc["population_sha256"]!=population["population_sha256"] or dispatch_doc["assessment_sha256"]!=hashlib.sha256(assessment_raw).hexdigest() or dispatch_doc["decision_sha256"]!=hashlib.sha256(decision_raw).hexdigest(): raise ValueError("dispatch receipt binding differs")
    dispatch_rows=dispatch_doc["rows"]
    if not isinstance(dispatch_rows,list) or len(dispatch_rows)!=len(bindings): raise ValueError("dispatch receipt count differs")
    dispatched=set()
    for row in dispatch_rows:
        if not isinstance(row,dict) or set(row)!={"action_id","idempotency_key","operation","accepted","compensation_id"}: raise ValueError("dispatch row schema differs")
        action=row["action_id"]
        if action in dispatched or action not in bindings or row["accepted"] is not True or row["idempotency_key"]!=bindings[action][5] or row["operation"]!=decisions[action]["final_disposition"] or row["compensation_id"]!=decisions[action]["compensation_id"]: raise ValueError("dispatch row binding differs")
        dispatched.add(action)
    if dispatched!=set(bindings): raise ValueError("dispatch complete membership differs")
    final_raw=verify_pair(a.final_readback,a.final_readback_bundle,plan["final_readback_signer_id"],bootstrap,"final readback");finals=index_rows(strict_object(final_raw,"final readback"),FINAL_FIELDS,FINAL_ROW,"aidefend.final-authoritative-readback.v2",population,bindings,"final readback")
    now=datetime.now(timezone.utc);compensations=set()
    for action in sorted(bindings):
        decision,final=decisions[action],finals[action]
        if final["final_disposition"]!=decision["final_disposition"] or digest(final["state_sha256"],f"{action}.state_sha256")!=decision["expected_state_sha256"]: raise ValueError(f"{action}: final disposition/state differs")
        text(final["provider_receipt_id"],f"{action}.provider_receipt_id");fresh(final["observed_at"],f"{action}.observed_at",now,plan)
        if decision["final_disposition"] in REMEDIATING:
            comp=text(final["compensation_id"],f"{action}.compensation_id")
            if comp!=decision["compensation_id"] or comp in compensations: raise ValueError(f"{action}: compensation binding/reuse differs")
            compensations.add(comp)
        elif final["compensation_id"] not in (None,""): raise ValueError(f"{action}: unexpected compensation ID")
        if decision["final_disposition"]=="ACCEPTED_EXCEPTION":
            if final["exception_approval_id"]!=decision["exception_approval_id"]: raise ValueError(f"{action}: exception approval differs")
        elif final["exception_approval_id"] not in (None,""): raise ValueError(f"{action}: unexpected exception approval")
    value={"schema_version":"aidefend.reconciliation-final.v2","outcome":"PASS","incident_id":plan["incident_id"],"tenant_id":plan["tenant_id"],"plan_sha256":plan_sha,"population_receipt_sha256":hashlib.sha256(verify_pair(a.population,a.population_bundle,"r007_stage",bootstrap,"action population replay")).hexdigest(),"population_sha256":population["population_sha256"],"assessment_sha256":hashlib.sha256(assessment_raw).hexdigest(),"decision_sha256":hashlib.sha256(decision_raw).hexdigest(),"dispatch_sha256":hashlib.sha256(dispatch_raw).hexdigest(),"final_readback_sha256":hashlib.sha256(final_raw).hexdigest(),"evaluated_at":now.isoformat().replace("+00:00","Z"),"complete_action_count":len(bindings)}
    print(json.dumps({"outcome":"PASS","final_receipt_sha256":publish(a.output,value,bootstrap),"bootstrap_sha256":bootstrap_sha},sort_keys=True))

def main()->None:
    p=argparse.ArgumentParser();p.add_argument("mode",choices=("dispatch","verify"))
    for name in ("plan","plan-bundle","population","population-bundle","assessment","assessment-bundle","decisions","decisions-bundle"):p.add_argument("--"+name,type=Path,required=True)
    p.add_argument("--dispatch-receipt",type=Path);p.add_argument("--dispatch-receipt-bundle",type=Path);p.add_argument("--final-readback",type=Path);p.add_argument("--final-readback-bundle",type=Path);p.add_argument("--output",type=Path,required=True);a=p.parse_args()
    if a.mode=="verify" and any(value is None for value in (a.dispatch_receipt,a.dispatch_receipt_bundle,a.final_readback,a.final_readback_bundle)):p.error("verify requires dispatch receipt and final readback payload/bundle pairs")
    (dispatch if a.mode=="dispatch" else verify)(a)

if __name__=="__main__":main()

set -euo pipefail
umask 077
common=(--plan policy/reconciliation-plan.json --plan-bundle policy/reconciliation-plan.json.sigstore.json --population evidence/action-population.json --population-bundle evidence/action-population.json.sigstore.json --assessment evidence/reconciliation-assessment.json --assessment-bundle evidence/reconciliation-assessment.json.sigstore.json --decisions evidence/reconciliation-decisions.json --decisions-bundle evidence/reconciliation-decisions.json.sigstore.json)
python recovery/reconcile_final.py dispatch "${common[@]}" --output evidence/reconciliation-dispatch.json
# A separate read-only provider collector now creates and signs final-authoritative-readback.json.
python recovery/reconcile_final.py verify "${common[@]}" \
  --dispatch-receipt evidence/reconciliation-dispatch.json --dispatch-receipt-bundle evidence/reconciliation-dispatch.json.sigstore.json \
  --final-readback evidence/final-authoritative-readback.json --final-readback-bundle evidence/final-authoritative-readback.json.sigstore.json \
  --output evidence/reconciliation-final.json

Every network and Cosign operation has an authenticated finite deadline and byte ceiling. Redirects, compressed/oversized responses, missing or extra actions, a reused compensation ID, wrong final-state digest, unsigned exception, stale readback, provider outage, signature failure, or partial dispatch prevents the create-once final receipt. Irreversible effects are recorded as compensated or formally accepted; they are never described as erased.

Open-source tools

PostgreSQL
Temporal
OpenTelemetry
Cosign

Source-available / open-weight tools

Camunda 8 Self-Managed (Camunda License 1.0; source-available)

Commercial / hosted tools

Temporal Cloud
AWS Step Functions
Azure Durable Functions
Google Cloud Workflows

Threat mappings

MITRE ATLAS

AML.T0101 Data Destruction via AI Agent Tool Invocation (authoritative readback and compensation recover or disposition restorable downstream mutations)

MAESTRO

Agent Tool Misuse (L7) (incident-time cancellation or compensation addresses effects already issued through misused tools)
Repudiation (L7) (complete action/provider reconciliation establishes the final disposition of each issued effect)

OWASP LLM Top 10 2026

LLM03:2026 Excessive Agency (compensation addresses external effects already issued with excessive authority)

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

ASI02:2026 Tool Misuse and Exploitation (downstream effects of exploited tools are reversed, compensated, or explicitly dispositioned)
ASI08:2026 Cascading Failures (issued effects are reconciled and compensated before unresolved divergence propagates further)

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

AISubtech-2.1.2 Capability and Permission Overreach (Remediates the measured reversible downstream effects; irreversible effects are only dispositioned)
AISubtech-4.3.4 Replay Exploitation (Remediates the measured reversible downstream effects; irreversible effects are only dispositioned)
AITech-12.1 Tool Exploitation (Remediates the measured reversible downstream effects; irreversible effects are only dispositioned)
AITech-14.2 Abuse of Delegated Authority (Remediates the measured reversible downstream effects; irreversible effects are only dispositioned)
AITech-2.1 Excessive Agency (Remediates the measured reversible downstream effects; irreversible effects are only dispositioned)

Google Secure AI Framework 2.0 - Risks

RA: Rogue Actions (external effects of rogue actions are reconciled and remediated after containment)

Databricks AI Security Framework 3.0

Agents - Core 13.2: Tool Misuse (downstream effects of misused tools are reconciled and remediated)
Agents - Core 13.8: Repudiation & Untraceability (ledger/provider reconciliation restores traceability and final disposition)