AID-I-004.004: Transactional Promotion Gates (Quarantine -> Trusted)

Version 1.20261015 · Actionable sub-technique

Implements a strict state machine and atomic promotion workflow for high-risk memory writes. Items routed into quarantine cannot influence agent behavior until reviewed and promoted. Promotion must be transactional, auditable, and typically re-signed as trusted. Aligns with trust-tiered memory write-gates (e.g., trusted/probation/quarantined).

Pillars: app, data

Lifecycle: operation

Scope boundary

Owns the durable memory state machine and transactional promotion of an exact quarantined write into trusted memory, including review binding, compare-and-swap state transition, re-signing, audit evidence, and idempotent publication. It does not define the upstream eligibility contract or authorize transient peer data to persist automatically.

AID-M-002.004: AID-I-004.004 executes the runtime quarantine-to-trusted promotion; AID-M-002.004 defines the write-eligibility, provenance, trust-tier, evidence, and namespace contract consumed by that gate. A recommended trust tier is not a promotion result, and a promotion gate must not invent eligibility metadata outside the signed contract.

AID-I-004.009: AID-I-004.004 promotes an approved write into durable trusted memory; AID-I-004.009 promotes a bounded peer-message projection only into transient active context as peer-untrusted data. A peer context-promotion receipt never authorizes persistence, trust-tier elevation, or durable-memory promotion.

AID-I-004.003: AID-I-004.004 decides and records durable trust promotion; AID-I-004.003 signs and verifies persistent-memory content and metadata integrity. Promotion must consume integrity evidence, but integrity alone does not determine that the record is eligible for trusted use.

Implementation guidance

AID-I-004.004-G001

Promote quarantined memory through a compare-and-swap state machine and idempotent transactional outbox, binding one approved content digest to one controlled-writer receipt without claiming cross-store atomicity.
State and transaction boundary

Quarantine and trusted memory are separate security domains. A PostgreSQL transaction can atomically reserve one promotion and create an outbox event, but it cannot atomically commit an external controlled writer. Use an idempotent saga: the database records PROMOTION_PENDING, the writer accepts one promotion_id, and a reconciler records the signed writer receipt. Retrieval excludes every quarantine state. A crash after the external write is resolved by requesting the same idempotency key, never by creating another trusted record.

PostgreSQL state and outbox
CREATE TABLE quarantine_items (
  item_id uuid PRIMARY KEY,
  row_version bigint NOT NULL DEFAULT 1,
  tenant_id text NOT NULL,
  content text NOT NULL,
  content_sha256 text NOT NULL,
  target_namespace text NOT NULL,
  state text NOT NULL CHECK (
    state IN ('PENDING','PROMOTION_PENDING','PROMOTED','REJECTED')
  ),
  promotion_id uuid UNIQUE,
  approved_by text,
  approval_receipt_sha256 text,
  trusted_record_id text,
  trusted_envelope_sha256 text
);

CREATE TABLE memory_promotion_outbox (
  promotion_id uuid PRIMARY KEY,
  item_id uuid NOT NULL REFERENCES quarantine_items(item_id),
  tenant_id text NOT NULL,
  content_sha256 text NOT NULL,
  target_namespace text NOT NULL,
  approval_receipt_jws text NOT NULL,
  state text NOT NULL CHECK (state IN ('READY','DELIVERING','COMPLETE')),
  attempt_count integer NOT NULL DEFAULT 0,
  next_attempt_at timestamptz NOT NULL DEFAULT now(),
  writer_receipt_sha256 text
);

CREATE INDEX promotion_outbox_ready
  ON memory_promotion_outbox (next_attempt_at)
  WHERE state IN ('READY','DELIVERING');
Reserve exactly one approved promotion
# File: memory/promotion/reserve.py
from uuid import uuid4


def reserve_promotion(
    conn,
    *,
    item_id: str,
    expected_row_version: int,
    verified_principal: dict,
    approval_verifier,
) -> dict:
    snapshot = conn.execute(
        "SELECT row_version, tenant_id, content_sha256, "
        "target_namespace, state FROM quarantine_items "
        "WHERE item_id = %s",
        (item_id,),
    ).fetchone()
    if snapshot is None:
        raise LookupError("quarantine item not found")
    (
        snapshot_version, snapshot_tenant, snapshot_digest,
        snapshot_namespace, snapshot_state,
    ) = snapshot
    if (
        snapshot_state != "PENDING"
        or snapshot_version != expected_row_version
        or snapshot_tenant != verified_principal["tenant_id"]
    ):
        raise PermissionError("quarantine snapshot is not promotable")
    promotion_id = str(uuid4())
    # Human/policy approval happens without holding a database row lock.
    approval = approval_verifier.request_verify_and_consume({
        "action": "PROMOTE_QUARANTINED_MEMORY",
        "promotion_id": promotion_id,
        "item_id": item_id,
        "tenant_id": snapshot_tenant,
        "content_sha256": snapshot_digest,
        "target_namespace": snapshot_namespace,
        "row_version": snapshot_version,
        "principal_subject": verified_principal["subject"],
    })
    with conn.transaction():
        item = conn.execute(
            "SELECT row_version, tenant_id, content_sha256, "
            "target_namespace, state FROM quarantine_items "
            "WHERE item_id = %s FOR UPDATE",
            (item_id,),
        ).fetchone()
        if item is None:
            raise LookupError("quarantine item not found")
        (
            row_version, tenant_id, content_sha256, target_namespace, state
        ) = item
        if (
            (row_version, tenant_id, content_sha256, target_namespace, state)
            != snapshot
        ):
            raise PermissionError("quarantine item changed after approval")
        updated = conn.execute(
            "UPDATE quarantine_items SET state = 'PROMOTION_PENDING', "
            "promotion_id = %s, approved_by = %s, "
            "approval_receipt_sha256 = %s, row_version = row_version + 1 "
            "WHERE item_id = %s AND row_version = %s AND state = 'PENDING'",
            (
                promotion_id, verified_principal["subject"],
                approval["receipt_sha256"], item_id, row_version,
            ),
        ).rowcount
        if updated != 1:
            raise RuntimeError("promotion compare-and-swap failed")
        conn.execute(
            "INSERT INTO memory_promotion_outbox "
            "(promotion_id, item_id, tenant_id, content_sha256, "
            "target_namespace, approval_receipt_jws, state) "
            "VALUES (%s,%s,%s,%s,%s,%s,'READY')",
            (
                promotion_id, item_id, tenant_id, content_sha256,
                target_namespace, approval["receipt_jws"],
            ),
        )
    return {
        "promotion_id": promotion_id,
        "item_id": item_id,
        "state": "PROMOTION_PENDING",
        "approval_receipt_sha256": approval["receipt_sha256"],
    }
Idempotent delivery and reconciliation
# File: memory/promotion/deliver.py
def deliver_one(conn, controlled_writer, evidence) -> dict | None:
    with conn.transaction():
        job = conn.execute(
            "SELECT promotion_id, item_id, tenant_id, content_sha256, "
            "target_namespace, approval_receipt_jws "
            "FROM memory_promotion_outbox "
            "WHERE state IN ('READY','DELIVERING') "
            "AND next_attempt_at <= now() "
            "ORDER BY next_attempt_at FOR UPDATE SKIP LOCKED LIMIT 1"
        ).fetchone()
        if job is None:
            return None
        (
            promotion_id, item_id, tenant_id, content_sha256,
            target_namespace, approval_receipt_jws,
        ) = job
        conn.execute(
            "UPDATE memory_promotion_outbox SET state = 'DELIVERING', "
            "attempt_count = attempt_count + 1 WHERE promotion_id = %s",
            (promotion_id,),
        )
        content = conn.execute(
            "SELECT content FROM quarantine_items WHERE item_id = %s "
            "AND promotion_id = %s AND state = 'PROMOTION_PENDING'",
            (item_id, promotion_id),
        ).fetchone()
        if content is None:
            raise RuntimeError("promotion source state no longer matches")

    writer = controlled_writer.write_and_verify({
        "idempotency_key": promotion_id,
        "tenant_id": tenant_id,
        "namespace": target_namespace,
        "content": content[0],
        "expected_content_sha256": content_sha256,
        "approval_receipt_jws": approval_receipt_jws,
    })
    if (
        writer["content_sha256"] != content_sha256
        or writer["promotion_id"] != promotion_id
    ):
        raise RuntimeError("controlled-writer receipt misbound")

    with conn.transaction():
        updated = conn.execute(
            "UPDATE quarantine_items SET state = 'PROMOTED', "
            "trusted_record_id = %s, trusted_envelope_sha256 = %s, "
            "row_version = row_version + 1 "
            "WHERE item_id = %s AND promotion_id = %s "
            "AND state = 'PROMOTION_PENDING'",
            (
                writer["record_id"], writer["envelope_sha256"],
                item_id, promotion_id,
            ),
        ).rowcount
        if updated != 1:
            raise RuntimeError("promotion finalization lost binding")
        conn.execute(
            "UPDATE memory_promotion_outbox SET state = 'COMPLETE', "
            "writer_receipt_sha256 = %s WHERE promotion_id = %s",
            (writer["receipt_sha256"], promotion_id),
        )
    return evidence.append_and_verify({
        "schema_version": "aidefend.memory-promotion.v1",
        "promotion_id": promotion_id,
        "item_id": item_id,
        "tenant_id": tenant_id,
        "content_sha256": content_sha256,
        "target_namespace": target_namespace,
        "trusted_record_id": writer["record_id"],
        "trusted_envelope_sha256": writer["envelope_sha256"],
        "writer_receipt_sha256": writer["receipt_sha256"],
        "outcome": "PASS",
    })
Verify safely

replay the state machine and reconcile quarantine, outbox, writer, and trusted-store records under duplicate, crash, timeout, and retry tests.

Open-source tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-I-004.004: Transactional Promotion Gates (Quarantine -> Trusted) | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service