AID-I-004.007: Task-Bounded Context Segmentation & Secret Demotion
Segment active context by task phase within a live session. At each verified phase transition, replace raw secrets, privileged intermediate results, temporary credentials, and internal-only instructions with short-lived scoped handles, redacted summaries, or non-retrievable references when the next phase no longer requires the original value. Bind the transition evidence to the task, old and new phase, policy version, affected context fields, and demotion result.
Pillars: app
Lifecycle: operation
Scope boundary
Owns intra-session task-phase segmentation and demotion of raw secrets, privileged intermediate results, temporary credentials, and internal-only instructions into scoped handles, redacted summaries, or non-retrievable references when the next phase no longer needs them.
AID-I-004.001: AID-I-004.007 changes sensitive context at a verified phase transition within one session; AID-I-004.001 isolates and limits the session's volatile context as a whole. Session isolation does not prove that a secret was demoted when its task need ended, while phase demotion does not establish cross-session or cross-tenant isolation.
AID-I-004.005: AID-I-004.007 demotes sensitive active context because the live task changed phase; AID-I-004.005 expires, downgrades, or deletes persistent records under retention and staleness policy. The triggers, stores, evidence populations, and cleanup semantics are different even when both reduce retained sensitive state.
AID-I-004.008: AID-I-004.007 decides which context must be removed or demoted at a phase transition; AID-I-004.008 validates the security integrity of the summary or compaction transformation used to represent remaining context. A valid compaction does not decide phase need, and a phase policy does not prove that a generated summary preserved constraints or removed secrets correctly.
Implementation guidance
- AID-I-004.007-G001: Define task phases explicitly and trigger deterministic context pruning or demotion whenever the agent transitions into a new phase.
- AID-I-004.007-G002: Store high-sensitivity values outside conversational context and inject only short-lived opaque handles into active memory.
- AID-I-004.007-G003: Separate active context into distinct memory lanes so system-only, credential-bearing, and user-visible context cannot freely mix during a live session.
AID-I-004.007-G001
Define task phases explicitly and trigger deterministic context pruning or demotion whenever the agent transitions into a new phase.
Security boundary
A transition token is a short-lived, single-use state-machine capability. Hash the exact token, bind that digest and a monotonic transition sequence into the signed lease and the atomic context-store compare-and-swap, and use authority-provided time rather than the process wall clock. Revalidate the lease before and after every secret demotion, immediately before CAS, after the store commits, and after the exact readback. An expired, revoked, stale, replayed, or unverifiable transition cannot produce PASS.
Apply a lease-bound monotonic transition
# File: runtime/phase_transition.py
from __future__ import annotations
import hashlib
import json
import math
from dataclasses import dataclass
from typing import Literal, Protocol
@dataclass(frozen=True)
class AuthenticatedTask:
tenant_id: str
task_id: str
principal_id: str
@dataclass(frozen=True)
class VerifiedTransition:
receipt_id: str
receipt_sha256: str
tenant_id: str
task_id: str
principal_id: str
from_phase: str
to_phase: str
expected_context_revision: int
transition_sequence: int
transition_token_sha256: str
lease_id: str
policy_sha256: str
expires_at_authority_unix: float
@dataclass(frozen=True)
class VerifiedTransitionLease:
state: Literal["ACTIVE", "EXPIRED", "REVOKED"]
lease_id: str
transition_sequence: int
transition_token_sha256: str
authority_time_unix: float
expires_at_authority_unix: float
receipt_sha256: str
@dataclass(frozen=True)
class PhaseRule:
allow_labels: frozenset[str]
remove_labels: frozenset[str]
demote_labels: frozenset[str]
allowed_next_phases: frozenset[str]
@dataclass(frozen=True)
class VerifiedPhasePolicy:
policy_version: str
policy_sha256: str
rules: dict[str, PhaseRule]
@dataclass(frozen=True)
class ContextItem:
item_id: str
classification: str
content: object
provenance_sha256: str
@dataclass(frozen=True)
class ContextSnapshot:
revision: int
phase: str
items: tuple[ContextItem, ...]
last_transition_sequence: int
last_transition_token_sha256: str | None
last_transition_expires_at_authority_unix: float | None
@dataclass(frozen=True)
class OpaqueHandle:
handle_id: str
tenant_id: str
task_id: str
source_item_sha256: str
scope: str
transition_sequence: int
transition_token_sha256: str
expires_at_authority_unix: float
broker_receipt_sha256: str
@dataclass(frozen=True)
class CommittedTransition:
revision: int
transition_sequence: int
transition_token_sha256: str
transition_expires_at_authority_unix: float
output_sha256: str
cas_constraint_sha256: str
commit_receipt_sha256: str
@dataclass(frozen=True)
class TransitionResult:
control_status: Literal["PASS", "FAIL", "INSUFFICIENT_DATA", "ERROR"]
finding_status: Literal["NO_FINDING", "FINDING", "UNKNOWN"]
snapshot: ContextSnapshot | None
committed_revision: int | None
final_lease_receipt_sha256: str | None
detail: str
class WorkflowAuthority(Protocol):
def verify_and_consume(
self, *, task: AuthenticatedTask, transition_token: str,
) -> VerifiedTransition: ...
def load_policy(self, policy_sha256: str) -> VerifiedPhasePolicy: ...
def revalidate_transition(
self, *, transition: VerifiedTransition, stage: str,
) -> VerifiedTransitionLease: ...
class TrustedContextStore(Protocol):
def read_verified(self, task: AuthenticatedTask) -> ContextSnapshot: ...
def compare_and_swap(
self,
*,
task: AuthenticatedTask,
expected_revision: int,
expected_last_transition_sequence: int,
from_phase: str,
to_phase: str,
items: tuple[ContextItem, ...],
transition_receipt_id: str,
transition_sequence: int,
transition_token_sha256: str,
transition_expires_at_authority_unix: float,
lease_authority_time_unix: float,
lease_receipt_sha256: str,
output_sha256: str,
cas_constraint_sha256: str,
) -> CommittedTransition: ...
def read_revision(
self, task: AuthenticatedTask, revision: int,
) -> ContextSnapshot: ...
class SecretBroker(Protocol):
def demote(
self,
*,
task: AuthenticatedTask,
item: ContextItem,
transition_receipt_id: str,
transition_sequence: int,
transition_token_sha256: str,
expires_at_authority_unix: float,
lease_receipt_sha256: str,
) -> OpaqueHandle: ...
def revoke_uncommitted(self, handle: OpaqueHandle, *, reason: str) -> None: ...
class TransitionHalt(Exception):
def __init__(self, control_status: str, finding_status: str, detail: str):
super().__init__(detail)
self.control_status = control_status
self.finding_status = finding_status
self.detail = detail
class CompareAndSwapRejected(Exception):
"""A verified store rejection, distinct from an operational store fault."""
ALLOWED_REASONS = frozenset({
"REVISION_CONFLICT",
"SEQUENCE_STALE",
"TOKEN_REPLAY",
"LEASE_EXPIRED",
})
def __init__(self, reason: str):
if reason not in self.ALLOWED_REASONS:
raise ValueError("unknown compare-and-swap rejection reason")
super().__init__(reason)
self.reason = reason
def canonical_sha256(value) -> str:
raw = json.dumps(
value, sort_keys=True, separators=(",", ":"), ensure_ascii=False,
allow_nan=False,
).encode("utf-8")
return hashlib.sha256(raw).hexdigest()
def apply_phase_transition(
*,
task: AuthenticatedTask,
transition_token: str,
authority: WorkflowAuthority,
context_store: TrustedContextStore,
secret_broker: SecretBroker,
) -> TransitionResult:
if not all((task.tenant_id, task.task_id, task.principal_id, transition_token)):
raise PermissionError("authenticated task or transition token is incomplete")
transition_token_sha256 = hashlib.sha256(
transition_token.encode("utf-8")
).hexdigest()
transition = authority.verify_and_consume(
task=task, transition_token=transition_token,
)
if (
transition.tenant_id != task.tenant_id
or transition.task_id != task.task_id
or transition.principal_id != task.principal_id
or transition.transition_token_sha256 != transition_token_sha256
or not all((
transition.receipt_id, transition.lease_id,
transition.from_phase, transition.to_phase,
))
or len(transition.receipt_sha256) != 64
or len(transition.policy_sha256) != 64
or isinstance(transition.expected_context_revision, bool)
or transition.expected_context_revision < 0
or isinstance(transition.transition_sequence, bool)
or transition.transition_sequence < 1
or not math.isfinite(transition.expires_at_authority_unix)
):
raise PermissionError("transition receipt binding is invalid")
last_lease: VerifiedTransitionLease | None = None
committed: CommittedTransition | None = None
issued_handles: list[OpaqueHandle] = []
def nonpass(status: str, finding: str, detail: str) -> TransitionResult:
return TransitionResult(
control_status=status,
finding_status=finding,
snapshot=None,
committed_revision=None if committed is None else committed.revision,
final_lease_receipt_sha256=(
None if last_lease is None else last_lease.receipt_sha256
),
detail=detail,
)
def revalidate(stage: str) -> VerifiedTransitionLease:
nonlocal last_lease
try:
lease = authority.revalidate_transition(
transition=transition, stage=stage,
)
except Exception as exc:
raise TransitionHalt(
"ERROR", "UNKNOWN",
f"transition authority unavailable at {stage}",
) from exc
if (
lease.lease_id != transition.lease_id
or lease.transition_sequence != transition.transition_sequence
or lease.transition_token_sha256 != transition_token_sha256
or lease.expires_at_authority_unix
!= transition.expires_at_authority_unix
or len(lease.receipt_sha256) != 64
or not math.isfinite(lease.authority_time_unix)
or not math.isfinite(lease.expires_at_authority_unix)
):
raise TransitionHalt(
"FAIL", "FINDING", f"transition lease differs at {stage}",
)
last_lease = lease
if (
lease.state != "ACTIVE"
or lease.authority_time_unix >= lease.expires_at_authority_unix
):
raise TransitionHalt(
"FAIL", "FINDING", f"transition lease is not active at {stage}",
)
return lease
try:
revalidate("after_authorization")
policy = authority.load_policy(transition.policy_sha256)
if policy.policy_sha256 != transition.policy_sha256:
raise TransitionHalt(
"FAIL", "FINDING", "phase-policy digest differs",
)
rule = policy.rules.get(transition.from_phase)
if rule is None or transition.to_phase not in rule.allowed_next_phases:
raise TransitionHalt(
"FAIL", "FINDING", "phase transition is not allowed",
)
if (
rule.allow_labels & rule.remove_labels
or rule.allow_labels & rule.demote_labels
or rule.remove_labels & rule.demote_labels
):
raise TransitionHalt(
"ERROR", "UNKNOWN", "phase-policy label populations overlap",
)
snapshot = context_store.read_verified(task)
if (
snapshot.revision != transition.expected_context_revision
or snapshot.phase != transition.from_phase
or isinstance(snapshot.last_transition_sequence, bool)
or snapshot.last_transition_sequence < 0
):
raise TransitionHalt(
"FAIL", "FINDING", "context revision or current phase changed",
)
if (
transition.transition_sequence <= snapshot.last_transition_sequence
or transition_token_sha256 == snapshot.last_transition_token_sha256
):
raise TransitionHalt(
"FAIL", "FINDING", "transition sequence or token is stale/replayed",
)
projected: list[ContextItem] = []
seen_ids: set[str] = set()
for item in snapshot.items:
if (
not item.item_id
or item.item_id in seen_ids
or not item.classification
or len(item.provenance_sha256) != 64
):
raise TransitionHalt(
"ERROR", "UNKNOWN",
"trusted context item is invalid or duplicated",
)
seen_ids.add(item.item_id)
if item.classification in rule.remove_labels:
continue
if item.classification in rule.demote_labels:
before_demote = revalidate(f"before_demote:{item.item_id}")
source_sha256 = canonical_sha256({
"item_id": item.item_id,
"classification": item.classification,
"content": item.content,
"provenance_sha256": item.provenance_sha256,
})
try:
handle = secret_broker.demote(
task=task,
item=item,
transition_receipt_id=transition.receipt_id,
transition_sequence=transition.transition_sequence,
transition_token_sha256=transition_token_sha256,
expires_at_authority_unix=(
transition.expires_at_authority_unix
),
lease_receipt_sha256=before_demote.receipt_sha256,
)
except Exception as exc:
raise TransitionHalt(
"ERROR", "UNKNOWN", "secret demotion failed",
) from exc
issued_handles.append(handle)
if (
handle.tenant_id != task.tenant_id
or handle.task_id != task.task_id
or handle.source_item_sha256 != source_sha256
or handle.transition_sequence != transition.transition_sequence
or handle.transition_token_sha256 != transition_token_sha256
or handle.expires_at_authority_unix
!= transition.expires_at_authority_unix
or not handle.handle_id
or not handle.scope
or len(handle.broker_receipt_sha256) != 64
):
raise TransitionHalt(
"FAIL", "FINDING",
"secret-broker handle binding differs",
)
revalidate(f"after_demote:{item.item_id}")
projected.append(ContextItem(
item_id=item.item_id,
classification="opaque_secret_handle",
content={
"handle_id": handle.handle_id,
"scope": handle.scope,
"transition_sequence": handle.transition_sequence,
"transition_token_sha256": (
handle.transition_token_sha256
),
"expires_at_authority_unix": (
handle.expires_at_authority_unix
),
"broker_receipt_sha256": handle.broker_receipt_sha256,
},
provenance_sha256=item.provenance_sha256,
))
continue
if item.classification in rule.allow_labels:
projected.append(item)
# Unknown or unlisted classifications are removed by default.
serializable = [
{
"item_id": item.item_id,
"classification": item.classification,
"content": item.content,
"provenance_sha256": item.provenance_sha256,
}
for item in projected
]
output_sha256 = canonical_sha256(serializable)
before_cas = revalidate("before_cas")
cas_constraint = {
"schema_version": "aidefend.phase-transition-cas.v2",
"task_id": task.task_id,
"expected_revision": snapshot.revision,
"expected_last_transition_sequence": (
snapshot.last_transition_sequence
),
"transition_sequence": transition.transition_sequence,
"transition_token_sha256": transition_token_sha256,
"transition_expires_at_authority_unix": (
transition.expires_at_authority_unix
),
"output_sha256": output_sha256,
}
cas_constraint_sha256 = canonical_sha256(cas_constraint)
try:
committed = context_store.compare_and_swap(
task=task,
expected_revision=snapshot.revision,
expected_last_transition_sequence=(
snapshot.last_transition_sequence
),
from_phase=transition.from_phase,
to_phase=transition.to_phase,
items=tuple(projected),
transition_receipt_id=transition.receipt_id,
transition_sequence=transition.transition_sequence,
transition_token_sha256=transition_token_sha256,
transition_expires_at_authority_unix=(
transition.expires_at_authority_unix
),
lease_authority_time_unix=before_cas.authority_time_unix,
lease_receipt_sha256=before_cas.receipt_sha256,
output_sha256=output_sha256,
cas_constraint_sha256=cas_constraint_sha256,
)
except CompareAndSwapRejected as exc:
raise TransitionHalt(
"FAIL", "FINDING",
f"atomic CAS rejected transition: {exc.reason}",
) from exc
except Exception as exc:
raise TransitionHalt(
"ERROR", "UNKNOWN",
"atomic context-store compare-and-swap failed",
) from exc
if (
committed.transition_sequence != transition.transition_sequence
or committed.transition_token_sha256 != transition_token_sha256
or committed.transition_expires_at_authority_unix
!= transition.expires_at_authority_unix
or committed.output_sha256 != output_sha256
or committed.cas_constraint_sha256 != cas_constraint_sha256
or len(committed.commit_receipt_sha256) != 64
):
raise TransitionHalt(
"ERROR", "UNKNOWN", "CAS receipt binding differs",
)
revalidate("after_commit")
readback = context_store.read_revision(task, committed.revision)
if (
readback.phase != transition.to_phase
or readback.items != tuple(projected)
or readback.last_transition_sequence != transition.transition_sequence
or readback.last_transition_token_sha256 != transition_token_sha256
or readback.last_transition_expires_at_authority_unix
!= transition.expires_at_authority_unix
or canonical_sha256([
{
"item_id": item.item_id,
"classification": item.classification,
"content": item.content,
"provenance_sha256": item.provenance_sha256,
}
for item in readback.items
]) != output_sha256
):
raise TransitionHalt(
"ERROR", "UNKNOWN", "phase-transition readback differs",
)
final_lease = revalidate("after_readback_before_return")
return TransitionResult(
control_status="PASS",
finding_status="NO_FINDING",
snapshot=readback,
committed_revision=committed.revision,
final_lease_receipt_sha256=final_lease.receipt_sha256,
detail="monotonic transition committed and read back under active lease",
)
except TransitionHalt as halt:
if committed is None:
for handle in reversed(issued_handles):
try:
secret_broker.revoke_uncommitted(
handle, reason=halt.detail,
)
except Exception:
pass
return nonpass(halt.control_status, halt.finding_status, halt.detail)
Executable regression fixtures
# File: tests/test_phase_transition.py
import hashlib
from runtime.phase_transition import (
AuthenticatedTask,
CommittedTransition,
ContextItem,
ContextSnapshot,
CompareAndSwapRejected,
OpaqueHandle,
PhaseRule,
VerifiedPhasePolicy,
VerifiedTransition,
VerifiedTransitionLease,
apply_phase_transition,
canonical_sha256,
)
TOKEN = "one-time-transition-token"
TOKEN_SHA = hashlib.sha256(TOKEN.encode("utf-8")).hexdigest()
TASK = AuthenticatedTask("tenant-a", "task-a", "principal-a")
TRANSITION = VerifiedTransition(
"receipt-1", "r" * 64, "tenant-a", "task-a", "principal-a",
"research", "execute", 4, 12, TOKEN_SHA, "lease-12", "p" * 64, 200.0,
)
SECRET = ContextItem("secret-1", "secret", {"value": "hidden"}, "s" * 64)
class Authority:
def __init__(self, expire_at=None):
self.expire_at = expire_at
self.calls = []
def verify_and_consume(self, **_):
return TRANSITION
def load_policy(self, _):
return VerifiedPhasePolicy("v7", "p" * 64, {
"research": PhaseRule(
frozenset(), frozenset(), frozenset({"secret"}),
frozenset({"execute"}),
),
})
def revalidate_transition(self, *, transition, stage):
self.calls.append(stage)
state = "EXPIRED" if stage == self.expire_at else "ACTIVE"
return VerifiedTransitionLease(
state, transition.lease_id, transition.transition_sequence,
transition.transition_token_sha256, 100.0 + len(self.calls),
transition.expires_at_authority_unix, "l" * 64,
)
class Broker:
def __init__(self):
self.demotions = []
self.revocations = []
def demote(self, *, task, item, transition_sequence,
transition_token_sha256, expires_at_authority_unix, **_):
source_sha = canonical_sha256({
"item_id": item.item_id,
"classification": item.classification,
"content": item.content,
"provenance_sha256": item.provenance_sha256,
})
handle = OpaqueHandle(
"handle-1", task.tenant_id, task.task_id, source_sha, "tool:send",
transition_sequence, transition_token_sha256,
expires_at_authority_unix, "b" * 64,
)
self.demotions.append(handle)
return handle
def revoke_uncommitted(self, handle, *, reason):
self.revocations.append((handle.handle_id, reason))
class Store:
def __init__(self, last_sequence=11, cas_exception=None):
self.snapshot = ContextSnapshot(
4, "research", (SECRET,), last_sequence, "old-token", 150.0,
)
self.cas_calls = []
self.cas_exception = cas_exception
def read_verified(self, _):
return self.snapshot
def compare_and_swap(self, **values):
self.cas_calls.append(values)
if self.cas_exception is not None:
raise self.cas_exception
assert values["expected_revision"] == self.snapshot.revision
assert values["expected_last_transition_sequence"] == (
self.snapshot.last_transition_sequence
)
assert values["transition_sequence"] > self.snapshot.last_transition_sequence
assert values["transition_token_sha256"] != (
self.snapshot.last_transition_token_sha256
)
assert values["lease_authority_time_unix"] < (
values["transition_expires_at_authority_unix"]
)
assert values["cas_constraint_sha256"] == canonical_sha256({
"schema_version": "aidefend.phase-transition-cas.v2",
"task_id": TASK.task_id,
"expected_revision": self.snapshot.revision,
"expected_last_transition_sequence": (
self.snapshot.last_transition_sequence
),
"transition_sequence": values["transition_sequence"],
"transition_token_sha256": values["transition_token_sha256"],
"transition_expires_at_authority_unix": (
values["transition_expires_at_authority_unix"]
),
"output_sha256": values["output_sha256"],
})
revision = self.snapshot.revision + 1
self.snapshot = ContextSnapshot(
revision, values["to_phase"], values["items"],
values["transition_sequence"], values["transition_token_sha256"],
values["transition_expires_at_authority_unix"],
)
return CommittedTransition(
revision, values["transition_sequence"],
values["transition_token_sha256"],
values["transition_expires_at_authority_unix"],
values["output_sha256"], values["cas_constraint_sha256"], "c" * 64,
)
def read_revision(self, _, revision):
assert revision == self.snapshot.revision
return self.snapshot
def run(*, expire_at=None, last_sequence=11, cas_exception=None):
authority = Authority(expire_at)
broker = Broker()
store = Store(last_sequence, cas_exception)
result = apply_phase_transition(
task=TASK, transition_token=TOKEN, authority=authority,
context_store=store, secret_broker=broker,
)
return result, authority, broker, store
def test_expiry_before_secret_demotion_cannot_commit_or_pass():
result, _, broker, store = run(expire_at="before_demote:secret-1")
assert result.control_status == "FAIL"
assert not broker.demotions and not store.cas_calls
def test_expiry_after_demotion_revokes_handle_and_never_calls_cas():
result, _, broker, store = run(expire_at="after_demote:secret-1")
assert result.control_status == "FAIL"
assert broker.demotions and broker.revocations and not store.cas_calls
def test_stale_sequence_cannot_commit_or_pass():
result, _, _, store = run(last_sequence=12)
assert result.control_status == "FAIL"
assert not store.cas_calls
def test_verified_cas_conflict_is_a_finding():
result, _, _, store = run(
cas_exception=CompareAndSwapRejected("REVISION_CONFLICT"),
)
assert store.cas_calls
assert result.control_status == "FAIL"
assert result.finding_status == "FINDING"
def test_store_outage_is_error_unknown_not_a_finding():
result, _, _, store = run(
cas_exception=ConnectionError("context store unavailable"),
)
assert store.cas_calls
assert result.control_status == "ERROR"
assert result.finding_status == "UNKNOWN"
def test_success_binds_monotonic_cas_and_revalidates_after_readback():
result, authority, _, store = run()
assert result.control_status == "PASS"
assert store.cas_calls[0]["transition_sequence"] == 12
assert store.cas_calls[0]["transition_token_sha256"] == TOKEN_SHA
assert store.cas_calls[0]["transition_expires_at_authority_unix"] == 200.0
assert authority.calls[-3:] == [
"before_cas", "after_commit", "after_readback_before_return",
]
Atomic-store requirement: the context store must verify the lease receipt itself and, in one serializable transaction, require the expected revision and prior sequence, transition_sequence > stored_sequence, an unseen token digest, and trusted store/authority time strictly before expires_at; then persist the new sequence, token digest, expiry, items digest, and commit receipt together. The adapter must translate only a verified revision conflict, stale sequence, token replay, or lease expiry into CompareAndSwapRejected; connection, timeout, serialization, driver, and unknown faults must remain operational exceptions and produce ERROR/UNKNOWN. Never rely only on the caller-supplied authority timestamp. Reconcile any post-commit non-PASS result by committed revision; do not retry the one-time token.
Action: Treat authority outage, expiry/revocation, stale sequence, replayed token, secret-broker mismatch, CAS conflict, or readback mismatch as non-PASS. Revoke only uncommitted broker handles; once CAS commits, preserve the ledger and escalate post-commit uncertainty instead of attempting an unsafe rollback.
AID-I-004.007-G002
Store high-sensitivity values outside conversational context and inject only short-lived opaque handles into active memory.
Security boundary
An opaque handle is not a renamed secret ID. It is a cryptographically random, one-time capability stored only as a keyed digest and bound to verified tenant, task, workflow phase, tool audience, secret version, signed policy, and expiry. The broker never returns raw secret material to the model. At the authorized tool boundary it resolves the secret reference through a secret manager, invokes the exact tool operation, atomically consumes the handle, and returns only the tool result plus signed receipt.
Broker ledger
CREATE TABLE secret_handle_ledger (
handle_digest text PRIMARY KEY,
tenant_id text NOT NULL,
task_id text NOT NULL,
task_revision text NOT NULL,
phase text NOT NULL,
audience text NOT NULL,
secret_reference text NOT NULL,
secret_version text NOT NULL,
policy_version text NOT NULL,
policy_sha256 text NOT NULL,
expires_at timestamptz NOT NULL,
state text NOT NULL CHECK (
state IN ('ACTIVE','RESERVED','CONSUMED','REVOKED')
),
reservation_id uuid,
operation_sha256 text,
issued_at timestamptz NOT NULL,
consumed_at timestamptz,
evidence_receipt_sha256 text
);
REVOKE ALL ON secret_handle_ledger FROM PUBLIC;
GRANT SELECT, INSERT, UPDATE ON secret_handle_ledger TO secret_handle_broker;Mint, consume, and phase-exit revocation
# File: context/secret_handle_broker.py
from __future__ import annotations
import hashlib
import hmac
import secrets
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Callable
from uuid import uuid4
@dataclass(frozen=True)
class SecretExecutor:
"""Required authenticated runtime adapter callables; no fallback implementation."""
invoke_and_verify: Callable[[str, str, str, dict, str, str], dict]
@dataclass(frozen=True)
class Evidence:
"""Required authenticated runtime adapter callables; no fallback implementation."""
reserve_and_verify: Callable[[dict], dict]
finalize_and_verify: Callable[[str, dict], dict]
@dataclass(frozen=True)
class VerifiedPrincipal:
tenant_id: str
task_id: str
workload_id: str
@dataclass(frozen=True)
class VerifiedWorkflowState:
tenant_id: str
task_id: str
task_revision: str
phase: str
receipt_sha256: str
@dataclass(frozen=True)
class WorkflowAuthority:
"""Authenticated workflow service; the request cannot supply phase."""
read_current: Callable[[VerifiedPrincipal], VerifiedWorkflowState]
def require_current_workflow(
authority: WorkflowAuthority, principal: VerifiedPrincipal
) -> VerifiedWorkflowState:
state = authority.read_current(principal)
if (
not isinstance(state, VerifiedWorkflowState)
or state.tenant_id != principal.tenant_id
or state.task_id != principal.task_id
or not state.task_revision
or not state.phase
or not state.receipt_sha256.startswith("sha256:")
):
raise PermissionError("workflow authority returned invalid task state")
return state
PEPPER = Path("/run/secrets/secret-handle-pepper").read_bytes()
if len(PEPPER) < 32:
raise RuntimeError("secret-handle pepper must contain at least 256 bits")
def handle_digest(raw_handle: str) -> str:
return "hmac-sha256:" + hmac.new(
PEPPER, raw_handle.encode("ascii"), hashlib.sha256
).hexdigest()
def canonical_operation_sha256(operation: dict) -> str:
import jcs
return "sha256:" + hashlib.sha256(
bytes(jcs.canonicalize(operation))
).hexdigest()
def mint_handle(
conn,
*,
principal: VerifiedPrincipal,
workflow_authority: WorkflowAuthority,
audience: str,
secret_reference: str,
secret_version: str,
signed_policy: dict,
) -> dict:
workflow = require_current_workflow(workflow_authority, principal)
phase = workflow.phase
if (
signed_policy.get("schema_version") != "aidefend.secret-handle-policy.v1"
or not signed_policy.get("policy_version")
or not signed_policy.get("policy_sha256")
or isinstance(signed_policy.get("handle_allocation_attempts"), bool)
or not isinstance(signed_policy.get("handle_allocation_attempts"), int)
or signed_policy["handle_allocation_attempts"] < 1
or phase not in signed_policy.get("phase_ttl_seconds", {})
or audience not in signed_policy.get("allowed_audiences", [])
):
raise PermissionError("verified signed handle policy rejected request")
ttl_seconds = signed_policy["phase_ttl_seconds"][phase]
if (
isinstance(ttl_seconds, bool)
or not isinstance(ttl_seconds, int)
or ttl_seconds < 1
):
raise PermissionError("invalid policy-derived handle lifetime")
now = datetime.now(timezone.utc)
expires_at = datetime.fromtimestamp(
now.timestamp() + ttl_seconds, timezone.utc
)
# The raw handle is returned once; only its keyed digest is stored.
for attempt in range(signed_policy["handle_allocation_attempts"]):
raw_handle = secrets.token_urlsafe(32)
digest = handle_digest(raw_handle)
inserted = conn.execute(
"INSERT INTO secret_handle_ledger "
"(handle_digest, tenant_id, task_id, task_revision, phase, audience, "
"secret_reference, secret_version, policy_version, policy_sha256, "
"expires_at, state, issued_at) "
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,'ACTIVE',%s) "
"ON CONFLICT DO NOTHING RETURNING handle_digest",
(
digest, principal.tenant_id, principal.task_id,
workflow.task_revision, phase, audience, secret_reference, secret_version,
signed_policy["policy_version"], signed_policy["policy_sha256"],
expires_at, now,
),
).fetchone()
if inserted:
return {
"opaque_handle": raw_handle,
"handle_digest": digest,
"expires_at": expires_at.isoformat(),
"task_revision": workflow.task_revision,
"workflow_receipt_sha256": workflow.receipt_sha256,
"policy_version": signed_policy["policy_version"],
"policy_sha256": signed_policy["policy_sha256"],
}
raise RuntimeError("could not allocate unique opaque handle")
def consume_handle(
conn,
*,
principal: VerifiedPrincipal,
raw_handle: str,
workflow_authority: WorkflowAuthority,
audience: str,
operation: dict,
expected_policy_version: str,
expected_policy_sha256: str,
executor: SecretExecutor,
evidence: Evidence,
) -> dict:
workflow = require_current_workflow(workflow_authority, principal)
current_phase = workflow.phase
digest = handle_digest(raw_handle)
operation_sha256 = canonical_operation_sha256(operation)
now = datetime.now(timezone.utc)
reservation_id = str(uuid4())
# Commit the one-time reservation before any external tool side effect.
with conn.transaction():
row = conn.execute(
"SELECT tenant_id, task_id, task_revision, phase, audience, secret_reference, "
"secret_version, policy_version, policy_sha256, expires_at, state "
"FROM secret_handle_ledger WHERE handle_digest = %s FOR UPDATE",
(digest,),
).fetchone()
if row is None:
raise PermissionError("opaque handle unknown")
(
tenant_id, task_id, task_revision, phase, stored_audience, secret_reference,
secret_version, policy_version, policy_sha256, expires_at, state,
) = row
if (
state != "ACTIVE"
or tenant_id != principal.tenant_id
or task_id != principal.task_id
or task_revision != workflow.task_revision
or phase != current_phase
or stored_audience != audience
or policy_version != expected_policy_version
or policy_sha256 != expected_policy_sha256
or now >= expires_at
):
raise PermissionError("handle expired, consumed, or context-misbound")
updated = conn.execute(
"UPDATE secret_handle_ledger SET state = 'RESERVED', "
"reservation_id = %s, operation_sha256 = %s "
"WHERE handle_digest = %s AND state = 'ACTIVE'",
(reservation_id, operation_sha256, digest),
).rowcount
if updated != 1:
raise RuntimeError("one-time handle reservation lost race")
reservation = evidence.reserve_and_verify({
"schema_version": "aidefend.secret-handle-use.v1",
"handle_digest": digest,
"reservation_id": reservation_id,
"tenant_id": tenant_id,
"task_id": task_id,
"task_revision": task_revision,
"workflow_receipt_sha256": workflow.receipt_sha256,
"phase": phase,
"audience": audience,
"secret_version": secret_version,
"operation_sha256": operation_sha256,
"policy_version": policy_version,
"policy_sha256": policy_sha256,
"state": "RESERVED",
})
# The executor must persist idempotency by reservation_id before invoking
# the downstream tool, so reconciliation cannot duplicate the effect.
executed = executor.invoke_and_verify(
secret_reference=secret_reference,
secret_version=secret_version,
audience=audience,
operation=operation,
operation_sha256=operation_sha256,
idempotency_key=reservation_id,
)
final = evidence.finalize_and_verify(
reservation["receipt_jws"],
{
"schema_version": "aidefend.secret-handle-use.v1",
"handle_digest": digest,
"reservation_id": reservation_id,
"operation_sha256": operation_sha256,
"executor_receipt_sha256": executed["receipt_sha256"],
"result_sha256": executed["result_sha256"],
"state": "CONSUMED",
},
)
with conn.transaction():
updated = conn.execute(
"UPDATE secret_handle_ledger SET state = 'CONSUMED', "
"consumed_at = %s, evidence_receipt_sha256 = %s "
"WHERE handle_digest = %s AND state = 'RESERVED' "
"AND reservation_id = %s AND operation_sha256 = %s",
(
datetime.now(timezone.utc), final["receipt_sha256"], digest,
reservation_id, operation_sha256,
),
).rowcount
if updated != 1:
raise RuntimeError("reserved handle finalization lost binding")
return {
"tool_result": executed["tool_result"],
"result_sha256": executed["result_sha256"],
"evidence_receipt_sha256": final["receipt_sha256"],
}
def revoke_phase_handles(
conn,
*,
principal: VerifiedPrincipal,
completed_phase: str,
) -> int:
result = conn.execute(
"UPDATE secret_handle_ledger SET state = 'REVOKED' "
"WHERE tenant_id = %s AND task_id = %s AND phase = %s "
"AND state = 'ACTIVE'",
(principal.tenant_id, principal.task_id, completed_phase),
)
return result.rowcountDeployment contract: construct VerifiedPrincipal only after mTLS/SPIFFE or OIDC verification and tenant membership mapping. The model may carry the raw opaque handle but cannot call the ledger, extend expiry, choose tenant/task/revision/phase, or broaden audience. Both mint and consume read the current phase from the authenticated workflow authority; a request field or identity attribute alone is not phase authority. SecretExecutor runs as a separate tool gateway identity, resolves the secret reference in Vault/KMS/secret manager, injects it only into the exact outbound operation, and returns a signed receipt without raw secret. Evidence reservations and finalizations are signed, append-only, and locally verified; use bounded timeouts and reconciliation for a crash between external execution and ledger commit.
AID-I-004.007-G003
Separate active context into distinct memory lanes so system-only, credential-bearing, and user-visible context cannot freely mix during a live session.
Authoritative lane boundary
Do not accept a caller-provided lane, tag, or item list. A server-owned context store records a closed source_kind and returns the complete task/phase population in one immutable, content-addressed point-in-time snapshot. Middleware derives lanes and tags from code, applies policy-owned TTLs, and exports only explicitly model-visible lanes. System instructions use a separate trusted instruction channel, temporary tool work stays internal, and every secret-shaped capability must be an opaque handle whose active tenant/task/phase/audience binding is read from the authoritative broker in AID-I-004.007-G002. Raw secret material is never a valid lane payload.
Strict lane exporter
# File: memory/lane_exporter.py
from __future__ import annotations
import hashlib
import json
import re
from dataclasses import dataclass
from typing import Callable, Mapping
SHA256 = re.compile(r"^sha256:[0-9a-f]{64}$")
HANDLE_DIGEST = re.compile(r"^hmac-sha256:[0-9a-f]{64}$")
OPAQUE_HANDLE = re.compile(r"^[A-Za-z0-9_-]{43,128}$")
LANES = {"user_visible", "system_only", "secret_handles", "tool_working"}
MODEL_VISIBLE_LANES = {"user_visible", "secret_handles"}
SOURCE_RULES = {
"user_request": ("user_visible", "user_request"),
"task_outcome": ("user_visible", "task_outcome"),
"redacted_summary": ("user_visible", "redacted_summary"),
"trusted_system_policy": ("system_only", "system_policy"),
"safety_invariant": ("system_only", "safety_invariant"),
"secret_broker_handle": ("secret_handles", "opaque_secret_handle"),
"tool_result_temp": ("tool_working", "tool_result_temp"),
"privileged_intermediate": ("tool_working", "privileged_intermediate"),
}
ROW_KEYS = {
"item_id", "tenant_id", "task_id", "task_revision", "phase",
"source_kind", "created_at_epoch", "payload",
}
TEXT_PAYLOAD_KEYS = {"text"}
HANDLE_PAYLOAD_KEYS = {"opaque_handle", "audience"}
HANDLE_RECEIPT_KEYS = {
"state", "tenant_id", "task_id", "task_revision", "phase", "audience",
"policy_sha256", "handle_digest", "expires_at_epoch", "receipt_sha256",
}
SNAPSHOT_KEYS = {"snapshot_revision", "rows"}
@dataclass(frozen=True)
class VerifiedPrincipal:
tenant_id: str
task_id: str
task_revision: str
phase: str
workflow_receipt_sha256: str
@dataclass(frozen=True)
class VerifiedLanePolicy:
schema_version: str
policy_version: str
policy_sha256: str
secret_handle_policy_sha256: str
allowed_source_kinds_by_phase: Mapping[str, frozenset[str]]
lane_ttl_seconds: Mapping[str, int]
secret_handle_audiences: frozenset[str]
max_items_per_export: int
max_payload_bytes: int
@dataclass(frozen=True)
class AuthoritativeContextStore:
"""One authenticated point-in-time snapshot; callers supply no item list."""
read_snapshot: Callable[[VerifiedPrincipal], dict[str, object]]
@dataclass(frozen=True)
class SecretHandleAuthority:
"""Read-only adapter to the G002 broker ledger; never returns a raw secret."""
verify_active: Callable[..., dict[str, object]]
@dataclass(frozen=True)
class TrustedClock:
now_epoch: Callable[[], int]
@dataclass(frozen=True)
class EvidenceWriter:
append_and_readback: Callable[[dict[str, object]], dict[str, str]]
def canonical_sha256(value: object) -> str:
payload = json.dumps(
value, sort_keys=True, separators=(",", ":"), allow_nan=False
).encode("utf-8")
return "sha256:" + hashlib.sha256(payload).hexdigest()
def validate_bootstrap(
principal: VerifiedPrincipal,
policy: VerifiedLanePolicy,
) -> None:
if (
policy.schema_version != "aidefend.context-lane-policy.v1"
or not policy.policy_version
or SHA256.fullmatch(policy.policy_sha256) is None
or SHA256.fullmatch(policy.secret_handle_policy_sha256) is None
or set(policy.lane_ttl_seconds) != LANES
or not principal.task_revision
or SHA256.fullmatch(principal.workflow_receipt_sha256) is None
or principal.phase not in policy.allowed_source_kinds_by_phase
or not policy.secret_handle_audiences
):
raise PermissionError("verified lane policy is incomplete or inapplicable")
for value in (
policy.max_items_per_export,
policy.max_payload_bytes,
*policy.lane_ttl_seconds.values(),
):
if isinstance(value, bool) or not isinstance(value, int) or value < 1:
raise PermissionError("lane policy limits must be positive integers")
allowed = policy.allowed_source_kinds_by_phase[principal.phase]
if not allowed or not set(allowed).issubset(SOURCE_RULES):
raise PermissionError("phase contains an unknown or empty source-kind set")
for value in (principal.tenant_id, principal.task_id, principal.phase):
if not isinstance(value, str) or not value:
raise PermissionError("verified principal is incomplete")
def validate_text_payload(payload: object, max_bytes: int) -> str:
if not isinstance(payload, dict) or set(payload) != TEXT_PAYLOAD_KEYS:
raise PermissionError("text lane payload has an unexpected schema")
text = payload["text"]
if not isinstance(text, str) or not text:
raise PermissionError("text lane payload is empty or not text")
if len(text.encode("utf-8")) > max_bytes:
raise PermissionError("text lane payload exceeds signed policy")
return text
def verify_handle(
*,
payload: object,
principal: VerifiedPrincipal,
policy: VerifiedLanePolicy,
handles: SecretHandleAuthority,
now_epoch: int,
) -> tuple[dict[str, object], str]:
if not isinstance(payload, dict) or set(payload) != HANDLE_PAYLOAD_KEYS:
raise PermissionError("secret lane accepts only an opaque handle and audience")
raw_handle = payload["opaque_handle"]
audience = payload["audience"]
if (
not isinstance(raw_handle, str)
or OPAQUE_HANDLE.fullmatch(raw_handle) is None
or not isinstance(audience, str)
or audience not in policy.secret_handle_audiences
):
raise PermissionError("opaque handle shape or audience is invalid")
receipt = handles.verify_active(
raw_handle=raw_handle,
tenant_id=principal.tenant_id,
task_id=principal.task_id,
task_revision=principal.task_revision,
phase=principal.phase,
audience=audience,
expected_policy_sha256=policy.secret_handle_policy_sha256,
)
if not isinstance(receipt, dict) or set(receipt) != HANDLE_RECEIPT_KEYS:
raise PermissionError("broker returned an unexpected handle receipt")
expected = {
"state": "ACTIVE",
"tenant_id": principal.tenant_id,
"task_id": principal.task_id,
"task_revision": principal.task_revision,
"phase": principal.phase,
"audience": audience,
"policy_sha256": policy.secret_handle_policy_sha256,
}
if any(receipt.get(key) != value for key, value in expected.items()):
raise PermissionError("handle is inactive or context-misbound")
expires_at = receipt["expires_at_epoch"]
if (
isinstance(expires_at, bool)
or not isinstance(expires_at, int)
or expires_at <= now_epoch
or HANDLE_DIGEST.fullmatch(str(receipt["handle_digest"])) is None
or SHA256.fullmatch(str(receipt["receipt_sha256"])) is None
):
raise PermissionError("handle expiry, digest, or broker receipt is invalid")
exported = {
"type": "opaque_secret_handle",
"opaque_handle": raw_handle,
"audience": audience,
"expires_at_epoch": expires_at,
"broker_receipt_sha256": receipt["receipt_sha256"],
}
return exported, str(receipt["handle_digest"])
def export_for_model(
*,
principal: VerifiedPrincipal,
policy: VerifiedLanePolicy,
store: AuthoritativeContextStore,
handles: SecretHandleAuthority,
clock: TrustedClock,
evidence: EvidenceWriter,
) -> dict[str, object]:
validate_bootstrap(principal, policy)
now_epoch = clock.now_epoch()
if isinstance(now_epoch, bool) or not isinstance(now_epoch, int):
raise RuntimeError("trusted clock returned an invalid epoch")
snapshot = store.read_snapshot(principal)
if not isinstance(snapshot, dict) or set(snapshot) != SNAPSHOT_KEYS:
raise PermissionError("authoritative snapshot schema is invalid")
rows = snapshot["rows"]
if (
not isinstance(rows, list)
or len(rows) > policy.max_items_per_export
):
raise PermissionError("authoritative snapshot population is invalid")
# A successful authenticated empty snapshot is a known-zero population.
# Store/read failure raises above and cannot become a zero-row PASS.
if any(not isinstance(row, dict) or set(row) != ROW_KEYS for row in rows):
raise PermissionError("context row has an unexpected schema")
observed_ids = [row["item_id"] for row in rows]
if (
any(not isinstance(item_id, str) or not item_id for item_id in observed_ids)
or len(set(observed_ids)) != len(observed_ids)
):
raise PermissionError("snapshot has an invalid or duplicate item ID")
expected_ids = sorted(observed_ids)
ordered_rows = sorted(rows, key=lambda item: item["item_id"])
snapshot_binding = {
"tenant_id": principal.tenant_id,
"task_id": principal.task_id,
"task_revision": principal.task_revision,
"phase": principal.phase,
"rows": ordered_rows,
}
snapshot_sha256 = canonical_sha256(snapshot_binding)
if snapshot["snapshot_revision"] != snapshot_sha256:
raise PermissionError("snapshot revision does not bind the exact population")
allowed_sources = policy.allowed_source_kinds_by_phase[principal.phase]
model_items: list[dict[str, object]] = []
dispositions: list[dict[str, object]] = []
for row in ordered_rows:
if (
row["tenant_id"] != principal.tenant_id
or row["task_id"] != principal.task_id
or row["task_revision"] != principal.task_revision
or row["phase"] != principal.phase
):
raise PermissionError("context row crossed tenant, task, or phase")
source_kind = row["source_kind"]
if not isinstance(source_kind, str) or source_kind not in allowed_sources:
raise PermissionError("source kind is not allowed in this phase")
lane, tag = SOURCE_RULES[source_kind]
created_at = row["created_at_epoch"]
if isinstance(created_at, bool) or not isinstance(created_at, int):
raise PermissionError("context creation time is invalid")
age = now_epoch - created_at
if age < 0:
raise PermissionError("context row is dated in the future")
if age >= policy.lane_ttl_seconds[lane]:
dispositions.append({
"item_id": row["item_id"], "lane": lane, "tag": tag,
"state": "PRUNED_EXPIRED",
})
continue
if lane == "secret_handles":
exported, handle_digest = verify_handle(
payload=row["payload"],
principal=principal,
policy=policy,
handles=handles,
now_epoch=now_epoch,
)
model_items.append({
"item_id": row["item_id"], "lane": lane, "tag": tag,
"payload": exported,
})
dispositions.append({
"item_id": row["item_id"], "lane": lane, "tag": tag,
"state": "EXPORTED_HANDLE",
"handle_digest": handle_digest,
"handle_audience": exported["audience"],
"handle_expires_at_epoch": exported["expires_at_epoch"],
"broker_receipt_sha256": exported["broker_receipt_sha256"],
})
continue
text = validate_text_payload(row["payload"], policy.max_payload_bytes)
if lane in MODEL_VISIBLE_LANES:
model_items.append({
"item_id": row["item_id"], "lane": lane, "tag": tag,
"payload": {"text": text},
})
state = "EXPORTED_TEXT"
else:
# System-only content is loaded by the trusted instruction channel.
# Tool-working content stays internal until separately redacted.
state = "HELD_OUT_OF_MODEL"
dispositions.append({
"item_id": row["item_id"], "lane": lane, "tag": tag, "state": state,
})
if sorted(item["item_id"] for item in dispositions) != expected_ids:
raise RuntimeError("not every authoritative item received a disposition")
model_items_sha256 = canonical_sha256(model_items)
event_body = {
"schema_version": "aidefend.context-lane-export.v1",
"tenant_id": principal.tenant_id,
"task_id": principal.task_id,
"task_revision": principal.task_revision,
"workflow_receipt_sha256": principal.workflow_receipt_sha256,
"phase": principal.phase,
"policy_version": policy.policy_version,
"policy_sha256": policy.policy_sha256,
"snapshot_revision": snapshot["snapshot_revision"],
"snapshot_sha256": snapshot_sha256,
"population_count": len(expected_ids),
"population_sha256": canonical_sha256(expected_ids),
"exported_item_ids": sorted(item["item_id"] for item in model_items),
"model_items_sha256": model_items_sha256,
"dispositions": dispositions,
}
event_sha256 = canonical_sha256(event_body)
event = evidence.append_and_readback(event_body)
if (
not isinstance(event, dict)
or set(event) != {"event_sha256", "receipt_sha256"}
or event.get("event_sha256") != event_sha256
or SHA256.fullmatch(str(event["receipt_sha256"])) is None
):
raise RuntimeError("lane-export evidence readback is invalid or misbound")
return {
"model_items": model_items,
"model_items_sha256": model_items_sha256,
"snapshot_revision": snapshot["snapshot_revision"],
"snapshot_sha256": snapshot_sha256,
"population_sha256": canonical_sha256(expected_ids),
"event_sha256": event_sha256,
"evidence_receipt_sha256": event["receipt_sha256"],
}Executable negative tests
# File: tests/test_lane_exporter.py
from copy import deepcopy
import pytest
from memory.lane_exporter import (
AuthoritativeContextStore,
EvidenceWriter,
SecretHandleAuthority,
TrustedClock,
VerifiedLanePolicy,
VerifiedPrincipal,
canonical_sha256,
export_for_model,
)
DIGEST = "sha256:" + "a" * 64
HANDLE_POLICY_DIGEST = "sha256:" + "d" * 64
RECEIPT = "sha256:" + "b" * 64
HANDLE_DIGEST = "hmac-sha256:" + "c" * 64
HANDLE = "A" * 43
PRINCIPAL = VerifiedPrincipal(
"tenant-1", "task-1", "task-revision-7", "execute", RECEIPT
)
POLICY = VerifiedLanePolicy(
schema_version="aidefend.context-lane-policy.v1",
policy_version="lanes/7",
policy_sha256=DIGEST,
secret_handle_policy_sha256=HANDLE_POLICY_DIGEST,
allowed_source_kinds_by_phase={
"execute": frozenset({
"user_request", "trusted_system_policy",
"secret_broker_handle", "tool_result_temp",
})
},
lane_ttl_seconds={
"user_visible": 600,
"system_only": 600,
"secret_handles": 120,
"tool_working": 30,
},
secret_handle_audiences=frozenset({"payments-api"}),
max_items_per_export=20,
max_payload_bytes=4096,
)
def row(item_id, source_kind, payload, created_at=990):
return {
"item_id": item_id,
"tenant_id": "tenant-1",
"task_id": "task-1",
"task_revision": "task-revision-7",
"phase": "execute",
"source_kind": source_kind,
"created_at_epoch": created_at,
"payload": payload,
}
def run(
rows, *, handle_overrides=None, snapshot_overrides=None,
evidence_overrides=None, observed_events=None,
):
ordered_rows = sorted(deepcopy(rows), key=lambda item: item["item_id"])
snapshot_binding = {
"tenant_id": PRINCIPAL.tenant_id,
"task_id": PRINCIPAL.task_id,
"task_revision": PRINCIPAL.task_revision,
"phase": PRINCIPAL.phase,
"rows": ordered_rows,
}
snapshot = {
"snapshot_revision": canonical_sha256(snapshot_binding),
"rows": ordered_rows,
}
snapshot.update(snapshot_overrides or {})
store = AuthoritativeContextStore(
read_snapshot=lambda principal: deepcopy(snapshot),
)
expected_handle = {
"state": "ACTIVE",
"tenant_id": "tenant-1",
"task_id": "task-1",
"task_revision": "task-revision-7",
"phase": "execute",
"audience": "payments-api",
"policy_sha256": HANDLE_POLICY_DIGEST,
"handle_digest": HANDLE_DIGEST,
"expires_at_epoch": 1100,
"receipt_sha256": RECEIPT,
}
expected_handle.update(handle_overrides or {})
handles = SecretHandleAuthority(
verify_active=lambda **kwargs: dict(expected_handle)
)
def append_and_readback(event):
if observed_events is not None:
observed_events.append(deepcopy(event))
readback = {
"event_sha256": canonical_sha256(event),
"receipt_sha256": RECEIPT,
}
readback.update(evidence_overrides or {})
return readback
return export_for_model(
principal=PRINCIPAL,
policy=POLICY,
store=store,
handles=handles,
clock=TrustedClock(now_epoch=lambda: 1000),
evidence=EvidenceWriter(
append_and_readback=append_and_readback
),
)
def test_exports_only_user_text_and_active_handle():
observed_events = []
result = run([
row("01", "user_request", {"text": "summarize the invoice"}),
row("02", "trusted_system_policy", {"text": "approved policy"}),
row("03", "secret_broker_handle", {
"opaque_handle": HANDLE, "audience": "payments-api"
}),
row("04", "tool_result_temp", {"text": "temporary"}, created_at=900),
], observed_events=observed_events)
assert [item["item_id"] for item in result["model_items"]] == ["01", "03"]
handle = result["model_items"][1]["payload"]
assert handle == {
"type": "opaque_secret_handle", "opaque_handle": HANDLE,
"audience": "payments-api", "expires_at_epoch": 1100,
"broker_receipt_sha256": RECEIPT,
}
assert result["snapshot_revision"] == result["snapshot_sha256"]
assert result["model_items_sha256"] == canonical_sha256(result["model_items"])
disposition = observed_events[0]["dispositions"][2]
assert disposition["handle_audience"] == "payments-api"
assert disposition["handle_expires_at_epoch"] == 1100
assert disposition["broker_receipt_sha256"] == RECEIPT
def test_rejects_caller_injected_lane_or_raw_secret_field():
injected_lane = row("01", "user_request", {"text": "hello"})
injected_lane["lane"] = "user_visible"
with pytest.raises(PermissionError):
run([injected_lane])
with pytest.raises(PermissionError):
run([row("01", "secret_broker_handle", {
"opaque_handle": HANDLE,
"audience": "payments-api",
"raw_secret": "must-not-enter-context",
})])
def test_rejects_snapshot_revision_or_population_mismatch():
with pytest.raises(PermissionError):
run([row("01", "user_request", {"text": "hello"})],
snapshot_overrides={"snapshot_revision": DIGEST})
with pytest.raises(PermissionError):
run([
row("01", "user_request", {"text": "first"}),
row("01", "user_request", {"text": "duplicate"}),
])
def test_rejects_evidence_event_digest_mismatch():
with pytest.raises(RuntimeError):
run([row("01", "user_request", {"text": "hello"})],
evidence_overrides={"event_sha256": DIGEST})
@pytest.mark.parametrize("overrides", [
{"state": "REVOKED"},
{"task_id": "another-task"},
{"audience": "wrong-service"},
{"expires_at_epoch": 1000},
{"receipt_sha256": "not-a-digest"},
])
def test_rejects_inactive_or_misbound_handle(overrides):
with pytest.raises(PermissionError):
run([
row("01", "secret_broker_handle", {
"opaque_handle": HANDLE, "audience": "payments-api"
})
], handle_overrides=overrides)Deployment contract: construct VerifiedPrincipal only after workload/user authentication and tenant membership mapping; construct VerifiedLanePolicy only after signature, schema, applicability, monotonic-version, and freshness checks, and bind its secret_handle_policy_sha256 to the accepted G002 broker policy. Construct the snapshot adapter only over an authenticated datastore transaction that returns every row visible at one revision; its content-addressed revision must bind the verified principal and exact ordered rows. The context store, handle authority, and append-only evidence writer run under server identities unavailable to the model. The evidence writer must independently read back the appended canonical body and return both its recomputed event digest and signed receipt digest. Run the tests above plus wrong-phase, future-time, expired-item, unknown-source, oversized-payload, broker-outage, and evidence-outage cases. A row held out of the model is still accounted for in the population receipt; it is not silently discarded or reclassified.
Open-source tools
- Valkey
- Redis 8+ AGPLv3 source build
- OpenTelemetry
- Open Policy Agent (OPA)
- Microsoft Presidio
Commercial / hosted tools
- Redis Enterprise
- Nightfall DLP
- Akeyless Secrets Management
Threat mappings
MITRE ATLAS
- AML.T0051 LLM Prompt Injection
- AML.T0080 AI Agent Context Poisoning
- AML.T0080.000 AI Agent Context Poisoning: Memory
- AML.T0080.001 AI Agent Context Poisoning: Thread
MAESTRO
- Agent Goal Manipulation (L7)
- Data Leakage (Cross-Layer)
OWASP LLM Top 10 2026
- LLM01:2026 Prompt Injection (separate instruction and data lanes reduce cross-boundary context injection)
- LLM02:2026 Sensitive Information Disclosure
- LLM08:2026 Hidden Context Exposure (secret demotion removes credentials and internal instructions from model-visible context)
OWASP ML Top 10 2023
- N/A
OWASP Top 10 for Agentic Applications 2026
- ASI01:2026 Agent Goal Hijack
- ASI03:2026 Identity and Privilege Abuse
- ASI06:2026 Memory & Context Poisoning
NIST Adversarial Machine Learning 2025
- NISTAML.015 Indirect Prompt Injection
- NISTAML.018 Prompt Injection
- NISTAML.036 Leaking information from user interactions
Cisco Integrated AI Security and Safety Framework
- AISubtech-14.1.1 Credential Theft (phase demotion and scoped secret handles restrict raw-secret disclosure)
- AISubtech-14.1.2 Insufficient Access Controls (secret-backed tool use requires current task, phase and audience authorization)
- AISubtech-4.3.4 Replay Exploitation (one-use transition tokens and secret handles constrain capability replay)
- AISubtech-8.2.3 Data Exfiltration via Agent Tooling (phase demotion and scoped secret handles restrict raw-secret disclosure)
- AISubtech-8.4.1 System LLM Prompt Leakage (no-longer-needed internal instructions are removed or held out of visible context)
- AITech-4.2 Context Boundary Attacks
- AITech-8.2 Data Exfiltration / Exposure (phase demotion and scoped secret handles restrict raw-secret disclosure)
- AITech-8.3 Information Disclosure (phase demotion and scoped secret handles restrict raw-secret disclosure)
- AITech-8.4 Prompt/Meta Extraction (no-longer-needed internal instructions are removed or held out of visible context)
Google Secure AI Framework 2.0 - Risks
- PIJ: Prompt Injection
- SDD: Sensitive Data Disclosure
- RA: Rogue Actions
Databricks AI Security Framework 3.0
- Agents - Core 13.1: Memory Poisoning
- Agents - Core 13.6: Intent Breaking & Goal Manipulation
- Agents - Tools MCP Server 13.19: Credential and Token Exposure
- Agents - Tools MCP Server 13.24: Context Spoofing and Manipulation
- Agents - Tools MCP Client 13.34: Session and State Management Failures