AID-E-005: Compromised Durable Application Session & Agent State Teardown
Eviction actions
- Delete scoped application session records without claiming bearer-token, authorization-server, or gateway-session revocation
- Purge tainted durable conversational and agent memory
- Remove unauthorized webhook and tool registrations
- Cancel unauthorized queued jobs and schedule registrations
Pillars: infra, app
Lifecycle: response
Scope boundary
Owns teardown of explicitly scoped durable application sessions, conversational or agent memory, webhook and tool registrations, nonterminal queue records, and schedule registrations after compromise. It remains applicable when no active runtime exists and never claims identity-plane revocation or active worker termination.
AID-E-001: AID-E-005 deletes durable application and agent state; AID-E-001 revokes credentials, authentication sessions, principals, and delegated grants. Application teardown does not prove identity-plane eviction, and revoked identity objects do not remove durable state that could reload or reschedule compromised behavior.
AID-E-002: AID-E-005 removes durable queue, schedule, session, memory, and registration records; AID-E-002 terminates active processes, runs, execution leases, jobs, pods, and containers. Deleting a durable job record does not terminate a worker that already acquired it, while runtime termination does not remove the record that can launch replacement work.
AID-R-003.002: AID-E-005 removes compromised durable conversational and agent state; AID-R-003.002 restores an authority-selected clean snapshot into a new isolated namespace. Teardown is an eviction result, while restoration must separately prove that it excludes tainted, cross-tenant, post-cutoff, secret-bearing, and credential-bearing state.
Implementation guidance
- AID-E-005-G001: Delete only explicitly scoped durable application session records linked to the compromise.
- AID-E-005-G002: Produce a separately signed owner receipt and independently replayable zero-state verification for the exact durable-state scope.
- AID-E-005-G003: Purge only exact signed durable conversational-memory records that could reload attacker-controlled state.
- AID-E-005-G004: Remove exact unauthorized webhook and tool registrations under a signed teardown manifest.
- AID-E-005-G005: Delete exact nonterminal durable job and schedule records under an exhaustive state taxonomy; route runtime-owned or leased work to E-002.
AID-E-005-G001
Before you begin
JWTs, API keys, refresh tokens, signing keys, and provider credentials are E-001; browser-local state requires a separate client response.
Production implementation
A least-privilege adapter selected from trusted configuration performs a consistent pre-read of every exact record/index, compares tenant/principal/version/digest, then atomically deletes the session records and their exact reverse indexes/refresh associations or aborts. It writes a nonsecret tombstone/action event keyed by session ID and nonce where replay prevention requires it. Missing-before-action is not silently treated as successful deletion; record its observed state and require verifier reconciliation.
Independent verification
A different read-only identity queries every declared primary/replica/session index using the exact IDs, requires zero effective records and no stale index path able to reload them, and sends each safely captured old session identifier to the real authentication/session gate, which must deny while a clean control session succeeds. Replica lag/convergence bounds come from signed policy.
AID-E-005-G002
Receipt contents
E-005 closes only with two separately signed canonical artifacts: an owner receipt for the exact durable-state actions and a fresh verification receipt produced by a read-only identity that cannot delete state. The owner receipt must bind the exact canonical action-log bytes. Neither artifact accepts a caller-asserted digest or pass status. The verifier must query every store and state class declared by the signed scope using trusted, version-pinned adapters; omitted stores are unmeasured and fail closed.
Canonical-byte rule: The JSON examples below show the logical schemas. Serialize each artifact as UTF-8 canonical JSON using the supplied canonical_json_bytes function before signing, hashing, storage, or verification. The recovery gate receives those exact bytes and recomputes both digests; a parsed object plus a caller-provided hash is not acceptable evidence.
{
"schema_version": "aidefend.e005-state-owner-receipt/v1",
"control": "AID-E-005",
"system_or_tenant": "customer-support-prod",
"incident_id": "INC-2026-0718-0042",
"release_id": "recovery-2026-0718.1",
"scope_manifest_sha256": "6d6f9cb8164f379df6e0d4dc97e7c8797806f7657fd44a4c44a9ada0c1c6c8b3",
"action_log_sha256": "1111111111111111111111111111111111111111111111111111111111111111",
"declared_state": [
{"store_id": "sessions-primary", "state_class": "server_session", "selector_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
{"store_id": "memory-primary", "state_class": "conversation_memory", "selector_sha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"}
],
"completed_at": "2026-07-10T12:00:00Z"
}{
"schema_version": "aidefend.e005-state-action-log/v1",
"control": "AID-E-005",
"system_or_tenant": "customer-support-prod",
"incident_id": "INC-2026-0718-0042",
"release_id": "recovery-2026-0718.1",
"scope_manifest_sha256": "6d6f9cb8164f379df6e0d4dc97e7c8797806f7657fd44a4c44a9ada0c1c6c8b3",
"actions": [
{"store_id": "sessions-primary", "state_class": "server_session", "selector_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "action": "delete", "records_affected": 2, "result": "completed", "completed_at": "2026-07-10T11:59:57Z"},
{"store_id": "memory-primary", "state_class": "conversation_memory", "selector_sha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "action": "purge", "records_affected": 4, "result": "completed", "completed_at": "2026-07-10T11:59:59Z"}
],
"completed_at": "2026-07-10T11:59:59Z"
}{
"schema_version": "aidefend.e005-state-verification/v1",
"control": "AID-E-005",
"system_or_tenant": "customer-support-prod",
"incident_id": "INC-2026-0718-0042",
"release_id": "recovery-2026-0718.1",
"scope_manifest_sha256": "6d6f9cb8164f379df6e0d4dc97e7c8797806f7657fd44a4c44a9ada0c1c6c8b3",
"owner_receipt_sha256": "2222222222222222222222222222222222222222222222222222222222222222",
"action_log_sha256": "1111111111111111111111111111111111111111111111111111111111111111",
"observations": [
{"store_id": "sessions-primary", "state_class": "server_session", "selector_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "records_remaining": 0},
{"store_id": "memory-primary", "state_class": "conversation_memory", "selector_sha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "records_remaining": 0}
],
"verified_at": "2026-07-10T12:02:00Z",
"expires_at": "2026-07-10T12:17:00Z"
}Verify exact bytes, coverage, and event order before handoff
# File: incident_response/validate_e005_evidence.py
from __future__ import annotations
import hashlib
import hmac
import json
from datetime import datetime, timezone
from typing import Any
OWNER_FIELDS = {
"schema_version", "control", "system_or_tenant", "incident_id",
"release_id", "scope_manifest_sha256", "action_log_sha256",
"declared_state", "completed_at",
}
ACTION_LOG_FIELDS = {
"schema_version", "control", "system_or_tenant", "incident_id",
"release_id", "scope_manifest_sha256", "actions", "completed_at",
}
VERIFICATION_FIELDS = {
"schema_version", "control", "system_or_tenant", "incident_id",
"release_id", "scope_manifest_sha256", "owner_receipt_sha256",
"action_log_sha256", "observations", "verified_at", "expires_at",
}
STATE_FIELDS = {"store_id", "state_class", "selector_sha256"}
ACTION_FIELDS = STATE_FIELDS | {
"action", "records_affected", "result", "completed_at",
}
OBSERVATION_FIELDS = STATE_FIELDS | {"records_remaining"}
def parse_time(value: object, field: str) -> datetime:
if not isinstance(value, str) or not value:
raise ValueError(f"{field} must be a non-empty timestamp")
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.tzinfo is None:
raise ValueError(f"{field} must include a timezone")
return parsed.astimezone(timezone.utc)
def require_sha256(value: object, field: str) -> str:
if not isinstance(value, str) or len(value) != 64:
raise ValueError(f"{field} must be a SHA-256 hex digest")
try:
int(value, 16)
except ValueError as exc:
raise ValueError(f"{field} must be a SHA-256 hex digest") from exc
if value.lower() == "0" * 64:
raise ValueError(f"{field} must not be the unresolved zero digest")
return value.lower()
def canonical_json_bytes(value: dict[str, Any]) -> bytes:
return (
json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
allow_nan=False,
)
+ "\n"
).encode("utf-8")
def load_canonical_object(
artifact_bytes: bytes,
expected_fields: set[str],
label: str,
) -> dict[str, Any]:
if not isinstance(artifact_bytes, bytes) or not artifact_bytes:
raise ValueError(f"{label} must be non-empty bytes")
try:
decoded = artifact_bytes.decode("utf-8", errors="strict")
value = json.loads(decoded)
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ValueError(f"{label} is not strict UTF-8 JSON") from exc
if not isinstance(value, dict) or set(value) != expected_fields:
raise ValueError(f"{label} schema differs")
expected_bytes = canonical_json_bytes(value)
if not hmac.compare_digest(artifact_bytes, expected_bytes):
raise ValueError(f"{label} is not canonical JSON bytes")
return value
def digest_bytes(value: bytes) -> str:
return hashlib.sha256(value).hexdigest()
def state_key(item: object, expected_fields: set[str], label: str) -> tuple[str, str, str]:
if not isinstance(item, dict) or set(item) != expected_fields:
raise ValueError(f"{label} schema differs")
values = tuple(item.get(field) for field in ("store_id", "state_class", "selector_sha256"))
if any(not isinstance(value, str) or not value for value in values[:2]):
raise ValueError(f"{label} store and state class are required")
selector = require_sha256(values[2], f"{label}.selector_sha256")
return values[0], values[1], selector
def validate_e005_pair(
owner_artifact_bytes: bytes,
verification_artifact_bytes: bytes,
action_log_bytes: bytes,
now: datetime,
) -> None:
if now.tzinfo is None:
raise ValueError("now must include a timezone")
now = now.astimezone(timezone.utc)
owner = load_canonical_object(
owner_artifact_bytes, OWNER_FIELDS, "owner receipt"
)
action_log = load_canonical_object(
action_log_bytes, ACTION_LOG_FIELDS, "owner action log"
)
verification = load_canonical_object(
verification_artifact_bytes, VERIFICATION_FIELDS, "verification receipt"
)
if owner["schema_version"] != "aidefend.e005-state-owner-receipt/v1":
raise ValueError("unexpected owner schema version")
if action_log["schema_version"] != "aidefend.e005-state-action-log/v1":
raise ValueError("unexpected action-log schema version")
if verification["schema_version"] != "aidefend.e005-state-verification/v1":
raise ValueError("unexpected verifier schema version")
if owner["control"] != "AID-E-005":
raise ValueError("wrong control binding")
for field in ("system_or_tenant", "incident_id", "release_id"):
if not isinstance(owner[field], str) or not owner[field]:
raise ValueError(f"owner.{field} is required")
common = (
"control", "system_or_tenant", "incident_id",
"release_id", "scope_manifest_sha256",
)
if any(owner[key] != action_log[key] for key in common):
raise ValueError("owner and action-log scope bindings differ")
if any(owner[key] != verification[key] for key in common):
raise ValueError("owner and verifier scope bindings differ")
require_sha256(owner["scope_manifest_sha256"], "scope_manifest_sha256")
owner_sha256 = digest_bytes(owner_artifact_bytes)
action_log_sha256 = digest_bytes(action_log_bytes)
if not hmac.compare_digest(
require_sha256(
verification["owner_receipt_sha256"],
"verification.owner_receipt_sha256",
),
owner_sha256,
):
raise ValueError("verification does not bind the canonical owner bytes")
owner_action_digest = require_sha256(
owner["action_log_sha256"], "owner.action_log_sha256"
)
verification_action_digest = require_sha256(
verification["action_log_sha256"],
"verification.action_log_sha256",
)
if not hmac.compare_digest(owner_action_digest, action_log_sha256):
raise ValueError("owner does not bind the canonical action-log bytes")
if not hmac.compare_digest(verification_action_digest, action_log_sha256):
raise ValueError("verification does not bind the canonical action-log bytes")
declared_items = owner["declared_state"]
observed_items = verification["observations"]
action_items = action_log["actions"]
if not isinstance(declared_items, list) or not isinstance(observed_items, list):
raise ValueError("state coverage must be represented as lists")
if not isinstance(action_items, list):
raise ValueError("actions must be represented as a list")
declared = [
state_key(item, STATE_FIELDS, f"declared_state[{index}]")
for index, item in enumerate(declared_items)
]
observed = [
state_key(item, OBSERVATION_FIELDS, f"observations[{index}]")
for index, item in enumerate(observed_items)
]
action_scope = [
state_key(item, ACTION_FIELDS, f"actions[{index}]")
for index, item in enumerate(action_items)
]
if not declared or len(declared) != len(set(declared)):
raise ValueError("declared durable-state coverage is empty or duplicated")
if set(declared) != set(observed) or len(observed) != len(set(observed)):
raise ValueError("durable-state verification coverage differs")
if set(declared) != set(action_scope) or len(action_scope) != len(set(action_scope)):
raise ValueError("owner action-log coverage differs")
for index, item in enumerate(observed_items):
remaining = item.get("records_remaining")
if type(remaining) is not int or remaining != 0:
raise RuntimeError(
f"declared durable state remains at observations[{index}]"
)
for index, item in enumerate(action_items):
if not isinstance(item.get("action"), str) or not item["action"]:
raise ValueError(f"actions[{index}].action is required")
affected = item.get("records_affected")
if isinstance(affected, bool) or not isinstance(affected, int) or affected <= 0:
raise ValueError(f"actions[{index}].records_affected is invalid")
if item.get("result") != "completed":
raise RuntimeError(f"actions[{index}] did not complete")
owner_completed_at = parse_time(owner["completed_at"], "owner.completed_at")
action_completed_at = parse_time(
action_log["completed_at"], "action_log.completed_at"
)
verified_at = parse_time(verification["verified_at"], "verification.verified_at")
expires_at = parse_time(verification["expires_at"], "verification.expires_at")
action_event_times = [
parse_time(item["completed_at"], f"actions[{index}].completed_at")
for index, item in enumerate(action_items)
]
if any(event_time > action_completed_at for event_time in action_event_times):
raise ValueError("action event occurs after action-log completion")
if action_completed_at > owner_completed_at:
raise ValueError("owner receipt predates action-log completion")
if verified_at < owner_completed_at:
raise ValueError("verification predates owner completion")
if owner_completed_at > now or verified_at > now:
raise ValueError("owner or verification timestamp is from the future")
if expires_at <= verified_at or expires_at <= now:
raise ValueError("verification expiry is invalid or stale")
Separation: Verify the signature over the canonical owner bytes with the state-remediation key and the signature over canonical verification bytes with a different read-only verifier key before calling validate_e005_pair. Pin both public keys in the recovery gate. The verifier image owns the store queries; the incident manifest may supply exact selectors but never executable query text or adapter code.
Fail semantics: Noncanonical bytes, signature failure, recomputed digest mismatch, action-log mismatch, scope mismatch, duplicate or missing action/observation, verification before owner completion, stale receipt, unavailable store, adapter error, a zero-row remediation action, or any nonzero remaining count blocks E-005. Credential and token evidence is rejected here and routed to AID-E-001; runtime process evidence is rejected and routed to AID-E-002.
Before you begin
Use when at least one E-005-owned durable state class is applicable and a recovery gate requires an owner receipt plus independently replayable zero-state proof.
AID-E-005-G003
Production implementation
Within a serializable transaction or equivalent conditional batch, lock/re-read each record, compare tenant/identity/version/digest, delete exactly authorized rows and exact derived index/cache entries, update conversation aggregates without touching clean records, and commit durable tombstones that the ingestion/memory writer enforces before reinsertion. Unknown/missing/mismatched rows abort rather than being counted as removed.
Independent verification
A separate read-only identity queries every declared authoritative/derived store by exact IDs and conversation/agent indexes, requires zero reloadable records, verifies tombstone revision at each ingestion consumer, and launches an isolated replacement-agent read using the exact conversation/agent context; the deleted state must not reappear while a clean memory control remains readable.
Concrete PostgreSQL memory-store path
This method applies when the authoritative agent memory uses a managed PostgreSQL table. Use the existing incident verifier to provide a signed exact tenant/record/revision/content-digest scope before invoking this function. Configure memory_records(tenant_id text, record_id text, revision bigint, content_sha256 text, content bytea) with a primary key on (tenant_id, record_id), and memory_tombstones(tenant_id text, record_id text, content_sha256 text, incident_id text) with the same primary key. The memory writer must consult tombstones under the same per-tenant transaction lock before insert/update; it must derive content_sha256 from the actual stored bytes, not an agent-supplied field. Existing stores use an equivalent approved adapter, not this schema blindly.
# File: response/evict_postgres_memory.py
def evict_memory(connection, *, tenant_id, incident_id, targets):
# connection is a least-privilege psycopg connection; targets come from the
# verified incident scope, never directly from an agent/tool request.
if not tenant_id or not incident_id or not targets:
raise ValueError("exact authenticated eviction scope required")
ids = [row["record_id"] for row in targets]
if len(set(ids)) != len(ids):
raise ValueError("duplicate eviction target")
with connection.transaction():
with connection.cursor() as cursor:
# Writers use this same lock before changing the tenant memory.
cursor.execute("SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))", (tenant_id,))
for target in sorted(targets, key=lambda row: row["record_id"]):
cursor.execute("""SELECT revision, content_sha256 FROM memory_records
WHERE tenant_id=%s AND record_id=%s FOR UPDATE""",
(tenant_id, target["record_id"]))
row = cursor.fetchone()
if row != (target["revision"], target["content_sha256"]):
raise RuntimeError("memory row missing or drifted; abort entire batch")
cursor.execute("""INSERT INTO memory_tombstones
(tenant_id, record_id, content_sha256, incident_id) VALUES (%s,%s,%s,%s)""",
(tenant_id, target["record_id"], target["content_sha256"], incident_id))
cursor.execute("DELETE FROM memory_records WHERE tenant_id=%s AND record_id=%s",
(tenant_id, target["record_id"]))
if cursor.rowcount != 1:
raise RuntimeError("eviction population differs")
return {"status": "PRIMARY_REMOVED", "tenant_id": tenant_id,
"incident_id": incident_id, "record_ids": sorted(ids)}A tombstone check rejects the same record ID or blocked content digest even if reintroduced under another ID. The shared writer lock prevents a concurrent reinsert from racing the transaction. This function proves only primary-store removal. Independently reconcile tombstones, project deletion to every cache/vector index, and use a read-only identity plus an isolated replacement agent to verify that no target memory is reloadable. Cache/replica failures remain incomplete, not PASS. Preserve evidence before deletion; repeat calls require a verified prior incident receipt and readback rather than treating missing rows as success. Replay wrong tenant, changed revision/digest, a missing member, transaction failure and concurrent reinsert.
AID-E-005-G004
Production implementation
A version-pinned registry adapter authenticates with workload identity/mTLS, re-fetches each exact object, compares class/owner/version/endpoint/capability digests, disables then conditionally deletes only authorized objects, invalidates exact registry caches, and writes a durable tombstone/audit event to block stale reconciliation. Revoke webhook signing secrets or tool credentials through E-001 when applicable; never copy bearer tokens into commands/evidence.
Independent verification
A separate read-only registry identity queries every declared replica/cache and requires each exact ID absent/disabled with the tombstone revision accepted. It sends a safe synthetic event or tool invocation through the normal dispatcher and proves no callback/tool/MCP route resolves, while an approved clean registration control works. A fake or manifest-supplied URL is never contacted.
AID-E-005-G005
When durable-job teardown applies
Use this guidance only for durable job records and recurring schedule registrations in the application's authoritative job store. It does not broadcast to workers, revoke an in-memory task, terminate a process, delete an orchestrator Job object, or cascade-delete pods. Every nonterminal database state must be classified by a separately signed, exhaustive state taxonomy that is proven identical to the taxonomy enforced by the live database schema. A runtime-owned state or any active lease aborts the transaction, leaves durable records intact, and routes that exact execution to AID-E-002.
Authorize exact records and bind the exhaustive taxonomy
{
"schema_version": "aidefend.e005-work-teardown/v3",
"control": "AID-E-005",
"system_or_tenant": "customer-support-prod",
"incident_id": "INC-2026-0718-0042",
"job_record_ids": ["job-01J2Y4ZWV8C6M0BQ9F7S3K2H1D"],
"schedule_record_ids": ["schedule-01J2Y3P6K5T9R8V7N4C1M0QW2E"],
"state_taxonomy_sha256": "3333333333333333333333333333333333333333333333333333333333333333"
}{
"schema_version": "aidefend.e005-job-state-taxonomy/v1",
"taxonomy_id": "jobs-primary",
"taxonomy_version": "2026-07-10.1",
"db_schema_version": "jobs-schema-2026-07-10",
"states": [
{"state": "cancelled", "terminal": true, "owner_control": "NONE", "e005_deletable": false},
{"state": "claimed", "terminal": false, "owner_control": "AID-E-002", "e005_deletable": false},
{"state": "failed", "terminal": true, "owner_control": "NONE", "e005_deletable": false},
{"state": "queued", "terminal": false, "owner_control": "AID-E-005", "e005_deletable": true},
{"state": "retry_wait", "terminal": false, "owner_control": "AID-E-005", "e005_deletable": true},
{"state": "running", "terminal": false, "owner_control": "AID-E-002", "e005_deletable": false},
{"state": "scheduled", "terminal": false, "owner_control": "AID-E-005", "e005_deletable": true},
{"state": "succeeded", "terminal": true, "owner_control": "NONE", "e005_deletable": false}
]
}Serialize both the manifest and taxonomy as UTF-8 canonical JSON (sorted keys, compact separators, one trailing newline) before signing. The runner verifies each exact file under a trust key pinned in its image and rejects noncanonical bytes.
The states above are an example, not a universal list. The production taxonomy must enumerate every state accepted by the database. The script rejects it unless the signed entries exactly equal the live taxonomy table and a validated, non-null composite foreign key forces every durable_jobs row to use that taxonomy ID, version, and state.
cosign verify-blob --key keys/e005-approver.pub \
--bundle work-teardown.json.sigstore.json work-teardown.json
cosign verify-blob --key keys/e005-state-taxonomy-authority.pub \
--bundle work-state-taxonomy.json.sigstore.json work-state-taxonomy.json
Delete only E-005-owned nonterminal durable state in one transaction
# File: incident_response/delete_durable_work.py
from __future__ import annotations
import argparse
import hashlib
import hmac
import json
import math
import os
import subprocess
import tempfile
from pathlib import Path
from typing import Any
import psycopg
MANIFEST_FIELDS = {
"schema_version", "control", "system_or_tenant", "incident_id",
"job_record_ids", "schedule_record_ids", "state_taxonomy_sha256",
}
TAXONOMY_FIELDS = {
"schema_version", "taxonomy_id", "taxonomy_version",
"db_schema_version", "states",
}
STATE_FIELDS = {
"state", "terminal", "owner_control", "e005_deletable",
}
JOB_STATE_FK = "durable_jobs_state_taxonomy_fk"
JOB_STATE_COLUMNS = [
"state_taxonomy_id", "state_taxonomy_version", "state",
]
TAXONOMY_KEY_COLUMNS = [
"taxonomy_id", "taxonomy_version", "state",
]
MANIFEST_TRUST_KEY = Path("/opt/aidefend/trust/e005-approver.pub")
TAXONOMY_TRUST_KEY = Path(
"/opt/aidefend/trust/e005-state-taxonomy-authority.pub"
)
RUNTIME_PROFILE_VERSION = os.environ["AIDEFEND_RUNTIME_PROFILE_VERSION"].strip()
RUNTIME_PROFILE_SHA256 = os.environ["AIDEFEND_RUNTIME_PROFILE_SHA256"].strip()
COMMAND_TIMEOUT_SECONDS = float(os.environ["E005_COMMAND_TIMEOUT_SECONDS"])
if (
not RUNTIME_PROFILE_VERSION
or len(RUNTIME_PROFILE_SHA256) != 64
or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")
or RUNTIME_PROFILE_SHA256 == "0" * 64
or not math.isfinite(COMMAND_TIMEOUT_SECONDS)
or COMMAND_TIMEOUT_SECONDS <= 0
):
raise RuntimeError("versioned E-005 runtime profile is invalid")
def reject_duplicate_keys(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError(f"duplicate signed JSON key: {key}")
value[key] = item
return value
def reject_nonfinite(value):
raise ValueError(f"non-finite signed JSON value: {value}")
def strict_json(raw: bytes, label: str):
try:
return json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=reject_duplicate_keys,
parse_constant=reject_nonfinite,
)
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ValueError(f"{label} is not strict UTF-8 JSON") from exc
def verified_signed_bytes(
payload_path: Path,
bundle_path: Path,
trust_key: str | Path,
timeout_seconds: float,
label: str,
) -> bytes:
payload = payload_path.read_bytes()
bundle = bundle_path.read_bytes()
if not payload or not bundle:
raise ValueError(f"{label} payload or bundle is empty")
with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_snapshot = root / "payload"
bundle_snapshot = root / "payload.sigstore.json"
payload_snapshot.write_bytes(payload)
bundle_snapshot.write_bytes(bundle)
os.chmod(payload_snapshot, 0o400)
os.chmod(bundle_snapshot, 0o400)
subprocess.run(
["cosign", "verify-blob", "--key", str(trust_key),
"--bundle", str(bundle_snapshot), str(payload_snapshot)],
check=True, capture_output=True, text=True, timeout=timeout_seconds,
)
verified = payload_snapshot.read_bytes()
verified_bundle = bundle_snapshot.read_bytes()
if verified != payload or verified_bundle != bundle:
raise OSError(f"{label} verified payload or bundle snapshot changed")
return verified
class ActiveRuntimeOwnership(RuntimeError):
pass
def verify_blob(path: Path, bundle: Path, trust_key: Path) -> bytes:
return verified_signed_bytes(
path, bundle, trust_key, COMMAND_TIMEOUT_SECONDS, str(path)
)
def canonical_json_bytes(value: dict[str, Any]) -> bytes:
return (
json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
allow_nan=False,
)
+ "\n"
).encode("utf-8")
def load_canonical_object(
raw: bytes,
expected_fields: set[str],
label: str,
) -> tuple[dict[str, Any], bytes]:
if not raw:
raise ValueError(f"{label} is empty")
try:
value = json.loads(raw.decode("utf-8", errors="strict"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ValueError(f"{label} is not strict UTF-8 JSON") from exc
if not isinstance(value, dict) or set(value) != expected_fields:
raise ValueError(f"{label} schema differs")
if not hmac.compare_digest(raw, canonical_json_bytes(value)):
raise ValueError(f"{label} is not canonical JSON bytes")
return value, raw
def require_sha256(value: object, field: str) -> str:
if not isinstance(value, str) or len(value) != 64:
raise ValueError(f"{field} must be a SHA-256 hex digest")
try:
int(value, 16)
except ValueError as exc:
raise ValueError(f"{field} must be a SHA-256 hex digest") from exc
if value.lower() == "0" * 64:
raise ValueError(f"{field} must not be the unresolved zero digest")
return value.lower()
def exact_ids(manifest: dict[str, Any], field: str) -> list[str]:
values = manifest.get(field)
if not isinstance(values, list):
raise ValueError(f"{field} must be a list")
if (
values != sorted(set(values))
or any(not isinstance(item, str) or not item or "*" in item for item in values)
):
raise ValueError(f"{field} must be a sorted unique exact-ID list")
return values
def validate_taxonomy(taxonomy: dict[str, Any]) -> dict[str, dict[str, Any]]:
if taxonomy["schema_version"] != "aidefend.e005-job-state-taxonomy/v1":
raise ValueError("unexpected job-state taxonomy schema")
for field in ("taxonomy_id", "taxonomy_version", "db_schema_version"):
if not isinstance(taxonomy[field], str) or not taxonomy[field]:
raise ValueError(f"taxonomy.{field} is required")
items = taxonomy["states"]
if not isinstance(items, list) or not items:
raise ValueError("taxonomy.states must be a non-empty list")
if any(not isinstance(item, dict) or set(item) != STATE_FIELDS for item in items):
raise ValueError("taxonomy state schema differs")
names = [item["state"] for item in items]
if (
names != sorted(set(names))
or any(not isinstance(name, str) or not name for name in names)
):
raise ValueError("taxonomy states must be sorted, unique, non-empty names")
for item in items:
if type(item["terminal"]) is not bool or type(item["e005_deletable"]) is not bool:
raise ValueError(f"taxonomy booleans are invalid for {item['state']}")
if item["terminal"]:
if item["owner_control"] != "NONE" or item["e005_deletable"]:
raise ValueError(f"terminal state has an eviction owner: {item['state']}")
elif item["owner_control"] == "AID-E-002":
if item["e005_deletable"]:
raise ValueError(f"runtime state is marked E-005 deletable: {item['state']}")
elif item["owner_control"] == "AID-E-005":
if not item["e005_deletable"]:
raise ValueError(f"E-005 state lacks its durable action: {item['state']}")
else:
raise ValueError(f"nonterminal state has no control owner: {item['state']}")
return {item["state"]: item for item in items}
def verify_database_taxonomy(
cursor: psycopg.Cursor,
taxonomy: dict[str, Any],
policy: dict[str, dict[str, Any]],
) -> None:
cursor.execute(
"""
SELECT
constraint_row.convalidated,
ARRAY(
SELECT attribute.attname
FROM unnest(constraint_row.conkey)
WITH ORDINALITY AS key_column(attnum, ordinal)
JOIN pg_attribute AS attribute
ON attribute.attrelid = constraint_row.conrelid
AND attribute.attnum = key_column.attnum
ORDER BY key_column.ordinal
),
ARRAY(
SELECT attribute.attname
FROM unnest(constraint_row.confkey)
WITH ORDINALITY AS key_column(attnum, ordinal)
JOIN pg_attribute AS attribute
ON attribute.attrelid = constraint_row.confrelid
AND attribute.attnum = key_column.attnum
ORDER BY key_column.ordinal
)
FROM pg_constraint AS constraint_row
WHERE constraint_row.conrelid = 'durable_jobs'::regclass
AND constraint_row.confrelid = 'durable_job_state_taxonomy'::regclass
AND constraint_row.conname = %s
AND constraint_row.contype = 'f'
""",
(JOB_STATE_FK,),
)
fk_rows = cursor.fetchall()
if len(fk_rows) != 1:
raise RuntimeError("required durable-job state taxonomy FK is missing")
validated, source_columns, target_columns = fk_rows[0]
if (
validated is not True
or list(source_columns) != JOB_STATE_COLUMNS
or list(target_columns) != TAXONOMY_KEY_COLUMNS
):
raise RuntimeError("durable-job state taxonomy FK differs or is not validated")
cursor.execute(
"""
SELECT attname, attnotnull
FROM pg_attribute
WHERE attrelid = 'durable_jobs'::regclass
AND attname = ANY(%s)
AND NOT attisdropped
ORDER BY attname
""",
(JOB_STATE_COLUMNS,),
)
nullability = cursor.fetchall()
if (
[row[0] for row in nullability] != sorted(JOB_STATE_COLUMNS)
or any(row[1] is not True for row in nullability)
):
raise RuntimeError("durable-job taxonomy binding columns must be NOT NULL")
cursor.execute(
"""
SELECT
db_schema_version, state, is_terminal,
owner_control, e005_deletable
FROM durable_job_state_taxonomy
WHERE taxonomy_id = %s AND taxonomy_version = %s
ORDER BY state
""",
(taxonomy["taxonomy_id"], taxonomy["taxonomy_version"]),
)
database_rows = cursor.fetchall()
expected_rows = [
(
taxonomy["db_schema_version"],
state,
item["terminal"],
item["owner_control"],
item["e005_deletable"],
)
for state, item in sorted(policy.items())
]
if database_rows != expected_rows:
raise RuntimeError(
"signed taxonomy does not exactly equal the database-enforced taxonomy"
)
def classify_job(
row: tuple,
taxonomy: dict[str, Any],
policy: dict[str, dict[str, Any]],
) -> dict[str, Any]:
state = row[1]
if (
row[4] != taxonomy["taxonomy_id"]
or row[5] != taxonomy["taxonomy_version"]
or state not in policy
):
raise RuntimeError(f"job {row[0]} is outside the verified state taxonomy")
return policy[state]
def delete_durable_work(
manifest_path: Path,
manifest_bundle_path: Path,
taxonomy_path: Path,
taxonomy_bundle_path: Path,
) -> dict[str, Any]:
manifest_bytes = verify_blob(
manifest_path, manifest_bundle_path, MANIFEST_TRUST_KEY
)
taxonomy_bytes = verify_blob(
taxonomy_path, taxonomy_bundle_path, TAXONOMY_TRUST_KEY
)
manifest, _ = load_canonical_object(
manifest_bytes, MANIFEST_FIELDS, "work teardown manifest"
)
taxonomy, taxonomy_bytes = load_canonical_object(
taxonomy_bytes, TAXONOMY_FIELDS, "job-state taxonomy"
)
if manifest["schema_version"] != "aidefend.e005-work-teardown/v3":
raise ValueError("unexpected work-teardown schema")
if manifest["control"] != "AID-E-005":
raise ValueError("wrong control binding")
for field in ("system_or_tenant", "incident_id"):
if not isinstance(manifest[field], str) or not manifest[field]:
raise ValueError(f"manifest.{field} is required")
taxonomy_sha256 = hashlib.sha256(taxonomy_bytes).hexdigest()
if not hmac.compare_digest(
require_sha256(
manifest["state_taxonomy_sha256"],
"manifest.state_taxonomy_sha256",
),
taxonomy_sha256,
):
raise ValueError("manifest does not bind the canonical signed taxonomy")
policy = validate_taxonomy(taxonomy)
tenant = manifest["system_or_tenant"]
job_ids = exact_ids(manifest, "job_record_ids")
schedule_ids = exact_ids(manifest, "schedule_record_ids")
if not job_ids and not schedule_ids:
raise ValueError("at least one exact durable record ID is required")
e005_states = sorted(
state
for state, item in policy.items()
if not item["terminal"]
and item["owner_control"] == "AID-E-005"
and item["e005_deletable"]
)
if not e005_states:
raise ValueError("signed taxonomy contains no E-005-owned durable state")
with psycopg.connect(os.environ["JOB_DATABASE_URL"]) as connection:
with connection.transaction():
# This must remain the first transaction statement. Serializable
# isolation protects the schedule-child predicate until commit.
connection.execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")
with connection.cursor() as cursor:
verify_database_taxonomy(cursor, taxonomy, policy)
cursor.execute(
"SELECT job_id, state, lease_owner, schedule_id, "
"state_taxonomy_id, state_taxonomy_version "
"FROM durable_jobs "
"WHERE system_or_tenant = %s AND job_id = ANY(%s) "
"FOR UPDATE",
(tenant, job_ids),
)
jobs = cursor.fetchall()
if sorted(row[0] for row in jobs) != job_ids:
raise RuntimeError(
"authorized durable job set is not fully measurable"
)
cursor.execute(
"SELECT schedule_id FROM durable_job_schedules "
"WHERE system_or_tenant = %s AND schedule_id = ANY(%s) "
"FOR UPDATE",
(tenant, schedule_ids),
)
schedules = sorted(row[0] for row in cursor.fetchall())
if schedules != schedule_ids:
raise RuntimeError(
"authorized durable schedule set is not fully measurable"
)
# No state filter is permitted here. Every child row is classified
# through the signed taxonomy proven exhaustive by the live FK.
cursor.execute(
"SELECT job_id, state, lease_owner, schedule_id, "
"state_taxonomy_id, state_taxonomy_version "
"FROM durable_jobs "
"WHERE system_or_tenant = %s "
"AND schedule_id = ANY(%s) "
"FOR UPDATE",
(tenant, schedule_ids),
)
schedule_children = cursor.fetchall()
job_policy = {
row[0]: classify_job(row, taxonomy, policy)
for row in jobs
}
child_policy = {
row[0]: classify_job(row, taxonomy, policy)
for row in schedule_children
}
nonterminal_children = [
row
for row in schedule_children
if not child_policy[row[0]]["terminal"]
]
undeclared = sorted(
{row[0] for row in nonterminal_children} - set(job_ids)
)
if undeclared:
raise RuntimeError(
"schedule has undeclared nonterminal job records: "
+ ",".join(undeclared)
)
measured = {row[0]: row for row in jobs}
measured.update({row[0]: row for row in nonterminal_children})
measured_policy = dict(job_policy)
measured_policy.update(
{
row[0]: child_policy[row[0]]
for row in nonterminal_children
}
)
runtime_owned = sorted(
job_id
for job_id, row in measured.items()
if row[2] is not None
or measured_policy[job_id]["owner_control"] == "AID-E-002"
)
if runtime_owned:
raise ActiveRuntimeOwnership(
"runtime-owned or leased work must be contained by "
"AID-E-002: " + ",".join(runtime_owned)
)
invalid_targets = sorted(
row[0]
for row in jobs
if job_policy[row[0]]["terminal"]
or job_policy[row[0]]["owner_control"] != "AID-E-005"
or not job_policy[row[0]]["e005_deletable"]
)
if invalid_targets:
raise RuntimeError(
"job records are not E-005-owned nonterminal state: "
+ ",".join(invalid_targets)
)
cursor.execute(
"DELETE FROM durable_jobs "
"WHERE system_or_tenant = %s "
"AND job_id = ANY(%s) "
"AND state_taxonomy_id = %s "
"AND state_taxonomy_version = %s "
"AND state = ANY(%s) "
"AND lease_owner IS NULL "
"RETURNING job_id",
(
tenant,
job_ids,
taxonomy["taxonomy_id"],
taxonomy["taxonomy_version"],
e005_states,
),
)
deleted_jobs = sorted(row[0] for row in cursor.fetchall())
if deleted_jobs != job_ids:
raise RuntimeError("durable job deletion set differs")
cursor.execute(
"DELETE FROM durable_job_schedules "
"WHERE system_or_tenant = %s "
"AND schedule_id = ANY(%s) "
"RETURNING schedule_id",
(tenant, schedule_ids),
)
deleted_schedules = sorted(row[0] for row in cursor.fetchall())
if deleted_schedules != schedule_ids:
raise RuntimeError("durable schedule deletion set differs")
return {
"schema_version": "aidefend.e005-work-owner-receipt/v2",
"control": "AID-E-005",
"system_or_tenant": tenant,
"incident_id": manifest["incident_id"],
"job_record_ids": deleted_jobs,
"schedule_record_ids": deleted_schedules,
"state_taxonomy_id": taxonomy["taxonomy_id"],
"state_taxonomy_version": taxonomy["taxonomy_version"],
"state_taxonomy_sha256": taxonomy_sha256,
"runtime_profile_version": RUNTIME_PROFILE_VERSION,
"runtime_profile_sha256": RUNTIME_PROFILE_SHA256,
}
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("manifest", type=Path)
parser.add_argument("manifest_bundle", type=Path)
parser.add_argument("taxonomy", type=Path)
parser.add_argument("taxonomy_bundle", type=Path)
args = parser.parse_args()
receipt = delete_durable_work(
args.manifest,
args.manifest_bundle,
args.taxonomy,
args.taxonomy_bundle,
)
print(json.dumps(receipt, sort_keys=True, separators=(",", ":")))
if __name__ == "__main__":
main()
Independent verification: After commit, a separate read-only database identity re-verifies the signed taxonomy digest, exact database taxonomy rows, validated composite foreign key, and non-null binding columns. It queries the same tenant and exact job/schedule IDs, then queries every child of each removed schedule without a state filter and classifies terminality from that verified taxonomy. It signs selector-bound zero counts under aidefend.e005-state-verification/v1. A database or schema error, taxonomy drift, missing record before action, undeclared nonterminal child, runtime-owned or leased row, nonzero post-action count, or signature/scope mismatch blocks E-005.
Action: Remove only exact E-005-owned nonterminal durable job and schedule rows under the signed exhaustive taxonomy. Never use worker control broadcasts, process signals, Kubernetes Job deletion, pod cascading, queue flushes, or runtime termination as E-005 evidence; those active-runtime actions remain exclusively in AID-E-002.
Open-source tools
- Valkey
- Redis Open Source 8+ (AGPL-3.0 option)
- redis-cli
- PostgreSQL (PostgreSQL License; authoritative durable-state store and transactional readback)
- Psycopg 3 (LGPL-3.0-only; PostgreSQL teardown and verification adapter)
Commercial / hosted tools
- Splunk SOAR (orchestrates signed durable-state teardown adapters; not an authoritative store)
- Palo Alto Networks Cortex XSOAR (orchestrates signed durable-state teardown adapters; not an authoritative store)
- Redis Enterprise
- Temporal Cloud
Threat mappings
MITRE ATLAS
- AML.T0051 LLM Prompt Injection (purging injected session state)
- AML.T0054 LLM Jailbreak (terminating jailbroken sessions)
- AML.T0080 AI Agent Context Poisoning (purging durable memory and thread records prevents poisoned context from reloading)
- AML.T0080.000 AI Agent Context Poisoning: Memory (purging poisoned agent memory)
- AML.T0080.001 AI Agent Context Poisoning: Thread (purging poisoned conversation threads)
- AML.T0091.001 Use Alternate Authentication Material: Web Session Cookie (deleting the scoped server-side session record stops cookie replay)
- AML.T0108 AI Agent (C2)
- AML.T0092 Manipulate User LLM Chat History (session purge clears manipulated chat history)
- AML.T0121 AI Agent Environment Reconstruction (tainted durable agent state is purged so an evicted agent cannot resume from it)
- AML.T0130 AI Agent Response Biasing (removal of poisoned durable memory prevents persistent recommendation instructions from reloading)
MAESTRO
- Agent Goal Manipulation (L7)
- Compromised Agents (L7)
OWASP LLM Top 10 2026
- LLM01:2026 Prompt Injection (durable-state teardown prevents injected context from reloading)
- LLM02:2026 Sensitive Information Disclosure (removing compromised retained state prevents renewed exposure)
- LLM05:2026 Data and Model Poisoning (teardown removes durable poisoned conversational or agent state)
OWASP ML Top 10 2023
- N/A
OWASP Top 10 for Agentic Applications 2026
- ASI06:2026 Memory & Context Poisoning (purging poisoned memory and context)
- ASI10:2026 Rogue Agents (removing durable state that can respawn a rogue agent)
- ASI08:2026 Cascading Failures (removing durable state that could rehydrate a cascading compromise)
- ASI01:2026 Agent Goal Hijack (purging poisoned goal and conversational state prevents rehydration)
- ASI02:2026 Tool Misuse and Exploitation (removing unauthorized durable tool registrations and queued work)
NIST Adversarial Machine Learning 2025
- NISTAML.018 Prompt Injection (purging injected session state)
- NISTAML.039 Compromising connected resources
- NISTAML.015 Indirect Prompt Injection (purging injected durable state prevents the chain from being reloaded)
- NISTAML.036 Leaking information from user interactions
Cisco Integrated AI Security and Safety Framework
- AISubtech-14.1.2 Insufficient Access Controls (Removes the compromised application session; identity-plane token revocation is separate)
- AISubtech-4.2.2 Session Boundary Violation (Evicts compromised durable session/context state; identity-plane credentials are separate)
- AISubtech-4.3.4 Replay Exploitation (Stops replay of the evicted application session identifiers)
- AISubtech-5.1.1 Long-term / Short-term Memory Injection
- AISubtech-9.3.1 Malicious Package / Tool Injection (Removes the identified malicious tool registration and all dispatch-cache routes)
- AITech-12.1 Tool Exploitation (Removes the identified malicious tool registration and all dispatch-cache routes)
- AITech-4.2 Context Boundary Attacks (Evicts compromised durable session/context state; identity-plane credentials are separate)
- AITech-5.1 Memory System Persistence
- AITech-7.2 Memory System Corruption
Google Secure AI Framework 2.0 - Risks
- PIJ: Prompt Injection (state purging removes injected durable session or memory state)
- SDD: Sensitive Data Disclosure (removing compromised retained session or conversational state limits subsequent disclosure)
- RA: Rogue Actions (state purging removes poisoned agent memory enabling rogue actions)
- IIC: Insecure Integrated Component (dismantling malicious webhooks and tool registrations)
Databricks AI Security Framework 3.0
- Agents - Core 13.1: Memory Poisoning
- Agents - Core 13.2: Tool Misuse
- Agents - Core 13.6: Intent Breaking & Goal Manipulation
- Agents - Core 13.13: Rogue Agents in Multi-Agent Systems
- Agents - Core 13.12: Agent Communication Poisoning
- Agents - Tools MCP Client 13.34: Session and State Management Failures