Scope boundary

Owns the semantics, canonical representation, authenticity, exact agent/release/authority binding, and effective-state reconciliation of the approved goal baseline across valid versions. Initial creation and later consumption use the same artifact and verifier.

AID-M-009.005: AID-M-003.004 proves which approved mission is currently effective and supplies its trustworthy meaning and digest; AID-M-009.005 proves that a transition from that state was authorized, retained, promoted, and can be recovered safely. One canonical manifest, signature, and replica readback may support both claims but receive no duplicate integrity or risk-reduction credit.

AID-M-009.002: AID-M-003.004 records intended mission and binds its authority-envelope digest; AID-M-009.002 defines concrete authority limits and action-risk facts. Mission text never expands the signed authority envelope.

AID-H-021.002: AID-M-003.004 verifies goal semantics and release-bound effective baseline; AID-H-021.002 owns generic fail-closed startup and reload admission of signed runtime artifacts. The goal-specific verifier is reused at that admission point; do not score the same signature rejection twice.

AID-D-010: AID-M-003.004 supplies the current verified goal and canonical digest; AID-D-010 detects semantic deviation of plans and actions. Valid goal provenance does not prove aligned behavior.

Implementation guidance

• AID-M-003.004-G001: Define the mission and constraint baseline using one canonical goal-manifest schema with typed genesis and predecessor binding.
• AID-M-003.004-G002: Sign and independently verify the exact canonical goal artifact with an authorized protected release identity.
• AID-M-003.004-G003: Attach the verified current mission semantics and exact goal digest to the agent or model card.
• AID-M-003.004-G004: Verify the signed active goal, prevent active-head replay, and reconcile its exact digest across the complete deployed replica population.

AID-M-003.004-G001

One goal artifact for the full lifecycle

Use this format for the initial mission, subsequent approved releases, and controlled restoration. Mission objectives and hierarchy describe intended behavior; authority limits come from the separately verified AID-M-009.002 envelope. Pin the Python runtime, pyca/cryptography and RFC8785 canonicalizer in a hash-locked release toolchain. The following concrete Ed25519 path is one implementation; GnuPG, Cosign or managed signers are valid when their adapters preserve the same canonical bytes, authorized issuer, digest and release binding.

# File: goal_contract.py
from __future__ import annotations
import base64, hashlib, json, re
from datetime import datetime, timezone
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
import rfc8785

SHA256 = re.compile(r"^[0-9a-f]{64}$")
GOAL_FIELDS = {
    "manifest_schema_version", "goal_id", "agent_id", "agent_release_digest",
    "mission_objectives", "goal_hierarchy", "allowed_tool_classes", "prohibited_actions",
    "data_and_tenant_boundaries", "human_approval_requirements", "authority_envelope_digest",
    "predecessor_goal_digest", "goal_generation", "effective_from", "expires_at",
}
HEAD_FIELDS = {"schema_version", "agent_id", "goal_manifest_digest", "agent_release_digest",
               "authority_envelope_digest", "goal_generation", "head_sequence", "activated_at", "expires_at"}

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

def canonical(value: dict) -> bytes:
    return rfc8785.dumps(value)

def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result: raise ValueError("duplicate JSON key")
        result[key] = value
    return result

def closed(value, fields, schema):
    if not isinstance(value, dict) or set(value) != fields:
        raise ValueError("closed schema fields differ")
    field = "manifest_schema_version" if "manifest_schema_version" in fields else "schema_version"
    if value[field] != schema: raise ValueError("schema version differs")

def text(value):
    if not isinstance(value, str) or not value.strip(): raise ValueError("empty or non-string identity")

def sha(value):
    if not isinstance(value, str) or not SHA256.fullmatch(value): raise ValueError("invalid SHA-256")

def population(value, *, empty=False):
    if not isinstance(value, list) or (not empty and not value): raise ValueError("missing population")
    for item in value: text(item)
    if len(value) != len(set(value)): raise ValueError("duplicate population")

def timestamp(value):
    text(value)
    t = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if t.tzinfo is None: raise ValueError("timestamp lacks timezone")
    return t.astimezone(timezone.utc)

def live_interval(start, end, now):
    if now.tzinfo is None: raise ValueError("clock lacks timezone")
    a, b = timestamp(start), timestamp(end)
    if not a <= now < b: raise ValueError("future, expired or invalid validity interval")

def positive_integer(value):
    if type(value) is not int or value < 1: raise ValueError("positive integer required")

def verify_signed(envelope, keys, allowed_key_ids):
    if not isinstance(envelope, dict) or set(envelope) != {"payload_b64", "signature_b64", "key_id"}:
        raise ValueError("signed envelope schema differs")
    kid = envelope["key_id"]
    if kid not in allowed_key_ids or kid not in keys: raise ValueError("signer is not authorized")
    key = keys[kid]
    if not isinstance(key, Ed25519PublicKey): raise ValueError("expected Ed25519 public key")
    raw = base64.b64decode(envelope["payload_b64"], validate=True)
    signature = base64.b64decode(envelope["signature_b64"], validate=True)
    key.verify(signature, raw)
    value = json.loads(raw.decode("utf-8"), object_pairs_hook=unique_object,
                       parse_constant=lambda x: (_ for _ in ()).throw(ValueError("non-finite JSON")))
    if not isinstance(value, dict) or canonical(value) != raw: raise ValueError("noncanonical signed payload")
    return value, digest(raw)

def validate_goal(goal, now):
    closed(goal, GOAL_FIELDS, "aidefend.agent-goal-manifest.v2")
    for field in ("goal_id", "agent_id"): text(goal[field])
    for field in ("agent_release_digest", "authority_envelope_digest"): sha(goal[field])
    for field in ("mission_objectives", "data_and_tenant_boundaries"):
        population(goal[field])
    for field in ("allowed_tool_classes", "prohibited_actions"):
        population(goal[field], empty=True)
    population(goal["human_approval_requirements"], empty=True)
    hierarchy = goal["goal_hierarchy"]
    if not isinstance(hierarchy, list) or not hierarchy: raise ValueError("missing goal hierarchy")
    names = []
    for row in hierarchy:
        if not isinstance(row, dict) or set(row) != {"objective", "sub_goals"}: raise ValueError("invalid hierarchy")
        text(row["objective"]); population(row["sub_goals"]); names.append(row["objective"])
    if len(names) != len(set(names)): raise ValueError("duplicate hierarchy objective")
    positive_integer(goal["goal_generation"])
    predecessor = goal["predecessor_goal_digest"]
    if goal["goal_generation"] == 1:
        if predecessor is not None: raise ValueError("genesis predecessor must be null")
    else:
        sha(predecessor)
    live_interval(goal["effective_from"], goal["expires_at"], now)

def validate_head(head, now):
    closed(head, HEAD_FIELDS, "aidefend.agent-goal-head.v2")
    text(head["agent_id"])
    for field in ("goal_manifest_digest", "agent_release_digest", "authority_envelope_digest"): sha(head[field])
    positive_integer(head["goal_generation"]); positive_integer(head["head_sequence"])
    live_interval(head["activated_at"], head["expires_at"], now)

def check_transition(candidate, current, current_digest):
    if current is None:
        if current_digest is not None or candidate["goal_generation"] != 1 or candidate["predecessor_goal_digest"] is not None:
            raise ValueError("invalid genesis transition")
        return
    sha(current_digest)
    if candidate["agent_id"] != current["agent_id"]: raise ValueError("cross-agent transition")
    if candidate["goal_generation"] != current["goal_generation"] + 1:
        raise ValueError("generation reset or skip")
    if candidate["predecessor_goal_digest"] != current_digest:
        raise ValueError("predecessor differs from the authoritative current goal")

Genesis and change admission

Generation one has a typed null predecessor. A later generation requires the exact current canonical goal digest and incremented generation. A null predecessor does not grant initialization: AID-M-009.005 also checks that the authoritative active-state store has no row for this agent. Missing store access is an error, never proof that the agent is new. Maintain complete declared objectives, tool classes, data/tenant boundaries and human approvals; compare them independently with the approved authority envelope and release inventory.

AID-M-003.004-G002

Sign the shared artifact

Load the candidate, current goal and active-head snapshot through the authenticated release pipeline. The protected signing service supplies the sign callback, while a separate verifier receives only the admission-pinned public keys and authorized key IDs. Never obtain trust roots, current-state absence, release identity or signer authorization from the agent, a pull request, or the candidate itself.

# File: sign_goal.py
import base64
from goal_contract import canonical, validate_goal, check_transition

def sign_goal(goal, *, now, current_goal, current_digest, protected_signer, key_id):
    validate_goal(goal, now)
    check_transition(goal, current_goal, current_digest)
    raw = canonical(goal)
    signature = protected_signer(raw)
    if not isinstance(signature, bytes) or len(signature) != 64: raise ValueError("invalid Ed25519 signature")
    return {"payload_b64": base64.b64encode(raw).decode("ascii"),
            "signature_b64": base64.b64encode(signature).decode("ascii"), "key_id": key_id}

Verify before publication

Call verify_signed and validate_goal over the returned envelope, compare its canonical digest with the signed deployment record, and publish immutable bytes plus signature together. Preserve the canonicalizer/runtime versions and exact key identity. AID-M-009.005 consumes this same envelope; creating a second signature or packaging wrapper earns no second integrity claim. Reject unknown fields, duplicate keys, wrong signer, noncanonical bytes, invalid predecessor type, future/expired validity and cross-release bindings.

AID-M-003.004-G003

Bind the descriptive card to verified goal bytes

Use the goal returned by the shared signature and active-head verifier, never a separately reopened YAML file or a digest of unverified content. The card remains descriptive metadata and cannot authorize actions or promotion.

# File: goal_card.py
from copy import deepcopy
from goal_contract import digest, canonical, validate_goal

def attach_goal_to_card(card, *, verified_goal, expected_goal_digest, now):
    validate_goal(verified_goal, now)
    if digest(canonical(verified_goal)) != expected_goal_digest: raise ValueError("goal digest differs")
    result = deepcopy(card)
    result["approved_agent_goal"] = {"agent_id": verified_goal["agent_id"],
        "release_digest": verified_goal["agent_release_digest"], "goal_manifest_digest": expected_goal_digest,
        "goal_generation": verified_goal["goal_generation"], "mission_objectives": verified_goal["mission_objectives"],
        "goal_hierarchy": verified_goal["goal_hierarchy"], "authority_envelope_digest": verified_goal["authority_envelope_digest"]}
    return result

def goal_monitor_input(verified_goal, *, expected_goal_digest, now):
    validate_goal(verified_goal, now)
    if digest(canonical(verified_goal)) != expected_goal_digest: raise ValueError("monitor goal binding differs")
    fragments = list(verified_goal["mission_objectives"])
    for row in verified_goal["goal_hierarchy"]:
        fragments.append(row["objective"]); fragments.extend(row["sub_goals"])
    return {"goal_digest": expected_goal_digest, "goal_text": "\n".join(fragments)}

Publish and read back

Write the card atomically in the protected release workspace, sign its final bytes, bind its digest to the deployment manifest, and read back the exact agent/release/goal fields after publication. Goal monitors use this canonical artifact through a pinned adapter that preserves objectives and hierarchy. Test both a genesis goal and an authorized successor against the same adapter.

AID-M-003.004-G004

Use the same verifier at the runtime handoff

Admission authenticates the deployment inventory, exact agent/release/authority tuple, goal-signing keys, active-head keys and per-instance readback signer authorization. Capture immutable envelopes before verification; fetch only from the fixed authenticated control-plane and digest-addressed artifact origin. Apply platform deadlines and byte limits from verified deployment policy.

# File: active_goal.py
from goal_store import open_goal_store
from goal_contract import verify_signed, validate_goal, validate_head, closed, sha, text, live_interval

def verify_active_goal(goal_envelope, head_envelope, *, keys, goal_keys, head_keys,
                       expected_agent, expected_release, expected_authority, now):
    goal, goal_digest = verify_signed(goal_envelope, keys, goal_keys)
    head, head_digest = verify_signed(head_envelope, keys, head_keys)
    validate_goal(goal, now); validate_head(head, now)
    for obj in (goal, head):
        if (obj["agent_id"], obj["agent_release_digest"], obj["authority_envelope_digest"]) != (
            expected_agent, expected_release, expected_authority): raise ValueError("deployment binding differs")
    if head["goal_manifest_digest"] != goal_digest or head["goal_generation"] != goal["goal_generation"]:
        raise ValueError("goal differs from signed active head")
    return goal, goal_digest, head, head_digest

def accept_head_floor(database, head, head_digest, *, busy_timeout_seconds, expected_store_id):
    if type(busy_timeout_seconds) not in (int, float) or not 0 < busy_timeout_seconds < float("inf"):
        raise ValueError("positive finite policy deadline required")
    with open_goal_store(database, expected_store_id=expected_store_id, kind="floor",
                         busy_timeout_seconds=busy_timeout_seconds) as conn:
        conn.execute("BEGIN IMMEDIATE")
        row = conn.execute("SELECT seq,digest FROM goal_head_floor WHERE agent=?", (head["agent_id"],)).fetchone()
        seq = head["head_sequence"]
        if row and (seq < row[0] or (seq == row[0] and head_digest != row[1])):
            raise ValueError("active-head rollback or sequence conflict")
        conn.execute("INSERT INTO goal_head_floor VALUES (?,?,?) ON CONFLICT(agent) DO UPDATE SET seq=excluded.seq,digest=excluded.digest",
                     (head["agent_id"], seq, head_digest))

def reconcile_replicas(envelopes, *, expected_instances, instance_keys, keys,
                       goal_digest, head_digest, agent_id, release_digest, challenge_nonce, now):
    if not expected_instances or len(expected_instances) != len(set(expected_instances)):
        raise ValueError("invalid expected replica population")
    text(challenge_nonce); seen = set()
    fields = {"schema_version", "instance_id", "agent_id", "release_digest", "goal_digest", "head_digest", "nonce", "issued_at", "expires_at"}
    for envelope in envelopes:
        # Select only an admission-pinned key; the signed subject is then checked
        # against the same inventory's per-instance signer authorization.
        row, _ = verify_signed(envelope, keys, set().union(*instance_keys.values()))
        closed(row, fields, "aidefend.goal-replica-readback.v2")
        instance = row["instance_id"]
        if instance not in expected_instances or instance in seen: raise ValueError("unexpected or duplicate replica")
        if envelope["key_id"] not in instance_keys.get(instance, ()): raise ValueError("unauthorized replica signer")
        if (row["agent_id"], row["release_digest"], row["goal_digest"], row["head_digest"], row["nonce"]) != (
                agent_id, release_digest, goal_digest, head_digest, challenge_nonce): raise ValueError("replica binding differs")
        live_interval(row["issued_at"], row["expires_at"], now); seen.add(instance)
    missing = sorted(set(expected_instances) - seen)
    return {"complete": not missing, "missing_instances": missing, "verified_instances": sorted(seen)}

Bootstrap and independent readback

Call verify_active_goal, then accept_head_floor on a protected persistent SQLite volume before exposing the goal to H-021.002 startup admission, H-018 authorization, or D-010 monitoring. Failures stop the handoff; no prompt, environment override or mutable alias may replace the verified goal. The head signer expires/revokes active state through the same authenticated publication path; stale but correctly signed artifacts cannot outrank the durable sequence floor.

Issue a fresh unpredictable readback nonce and collect independently signed envelopes from every inventory-listed replica; call reconcile_replicas with that exact population and per-instance authorized keys. Missing replicas remain insufficient evidence; wrong digest, release, head, nonce, signer, duplicate or unexpected replica is a finding. Preserve immutable input digests, trust configuration, head sequence, expected/observed populations and readback receipts. Signature-valid source bytes alone do not prove every runtime loaded them. These goal-specific facts do not earn a second score for generic runtime signature admission.

Provision state once under the independent control-plane administrator

Save the shared module below as goal_store.py beside active_goal.py and goal_transition.py. The mounted database directory must already exist. Only the separate provisioning authority initializes a store with the deployment-pinned store ID and role; service startup and runtime goal requests never provision tables or directories. The existing-only SQLite URI open rejects a missing volume or store atomically. Wrong store identity, role, schema, durability mode or empty database stops activation or baseline acceptance. Protect the mounted parent from agent/tool writes and store replacement, and keep backups plus independent signed head/retention authority outside the runtime identity.

# File: goal_store.py
from __future__ import annotations
import argparse
from contextlib import contextmanager
import math
import os
from pathlib import Path
import sqlite3

SCHEMA_VERSION = "aidefend.goal-store.v1"
TABLES = {
    "active": ("active_goals", "CREATE TABLE active_goals (agent TEXT PRIMARY KEY, seq INTEGER, goal_digest TEXT, head_digest TEXT, head_envelope TEXT)",
               [("agent", "TEXT"), ("seq", "INTEGER"), ("goal_digest", "TEXT"), ("head_digest", "TEXT"), ("head_envelope", "TEXT")]),
    "floor": ("goal_head_floor", "CREATE TABLE goal_head_floor (agent TEXT PRIMARY KEY, seq INTEGER, digest TEXT)",
              [("agent", "TEXT"), ("seq", "INTEGER"), ("digest", "TEXT")]),
}

def parameters(path, store_id, kind, deadline):
    if not isinstance(store_id, str) or not store_id or kind not in TABLES:
        raise ValueError("pinned store identity and role required")
    if type(deadline) not in (int, float) or not math.isfinite(deadline) or deadline <= 0:
        raise ValueError("positive finite policy deadline required")
    path = Path(path).absolute()
    if path.is_symlink(): raise PermissionError("goal store must not be a symlink")
    return path

def provision(path, *, store_id, kind, deadline):
    """Administrative initial provisioning only; never called by runtime."""
    path = parameters(path, store_id, kind, deadline)
    # The separately administered mounted directory must already exist. Never
    # create directories or overwrite a store in response to a runtime request.
    fd = os.open(path, os.O_RDWR | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o600)
    try: os.fsync(fd)
    finally: os.close(fd)
    conn = sqlite3.connect(path.as_uri() + "?mode=rw", uri=True, timeout=deadline)
    try:
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute("PRAGMA synchronous=FULL")
        with conn:
            conn.execute("CREATE TABLE goal_store_metadata (schema_version TEXT, store_id TEXT, kind TEXT)")
            conn.execute("INSERT INTO goal_store_metadata VALUES (?,?,?)", (SCHEMA_VERSION, store_id, kind))
            conn.execute(TABLES[kind][1])
            conn.execute("PRAGMA user_version=1")
    finally: conn.close()

@contextmanager
def open_goal_store(path, *, expected_store_id, kind, busy_timeout_seconds):
    path = parameters(path, expected_store_id, kind, busy_timeout_seconds)
    # mode=rw is an atomic existing-only open, not exists() followed by an
    # auto-creating connection. Missing mounts/stores cannot turn into genesis.
    conn = sqlite3.connect(path.as_uri() + "?mode=rw", uri=True,
                           timeout=busy_timeout_seconds)
    try:
        table, _sql, columns = TABLES[kind]
        names = {row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'")}
        if names != {table, "goal_store_metadata"}: raise PermissionError("goal store schema differs")
        if conn.execute("PRAGMA user_version").fetchone()[0] != 1:
            raise PermissionError("goal store schema version differs")
        metadata = conn.execute("SELECT schema_version,store_id,kind FROM goal_store_metadata").fetchall()
        if metadata != [(SCHEMA_VERSION, expected_store_id, kind)]:
            raise PermissionError("goal store identity or role differs")
        if [(row[1], row[2]) for row in conn.execute("PRAGMA table_info(" + table + ")")] != columns:
            raise PermissionError("goal store table columns differ")
        if conn.execute("PRAGMA journal_mode").fetchone()[0].lower() != "wal":
            raise PermissionError("goal store durability mode differs")
        conn.execute("PRAGMA synchronous=FULL")
        with conn: yield conn
    finally: conn.close()

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--provision", type=Path, required=True)
    parser.add_argument("--store-id", required=True)
    parser.add_argument("--kind", choices=TABLES, required=True)
    parser.add_argument("--deadline-seconds", type=float, required=True)
    args = parser.parse_args()
    provision(args.provision, store_id=args.store_id, kind=args.kind, deadline=args.deadline_seconds)

# Run once as the separate provisioning administrator, after verifying the mount.
python3 goal_store.py --provision /var/lib/aidefend/goals/active.sqlite3 \
  --store-id "$VERIFIED_ACTIVE_GOAL_STORE_ID" --kind active \
  --deadline-seconds "$VERIFIED_GOAL_STORE_PROVISION_DEADLINE_SECONDS"
python3 goal_store.py --provision /var/lib/aidefend/goals/head-floor.sqlite3 \
  --store-id "$VERIFIED_GOAL_HEAD_FLOOR_STORE_ID" --kind floor \
  --deadline-seconds "$VERIFIED_GOAL_STORE_PROVISION_DEADLINE_SECONDS"

Pass expected_store_id from the verified deployment configuration to both runtime functions; it is never selected by the agent or transition caller. Recover a lost store from the independently retained latest signed head and authoritative sequence; do not treat a fresh empty store as recovery or rerun genesis for an existing agent. Existing unversioned example databases require an administrator to preserve their current rows and reconcile them against that authority before adding the pinned metadata; do not migrate by deleting or replacing the history.

The provisioning administrator supplies the finite positive --deadline-seconds from the verified versioned storage/deployment policy, just as runtime supplies its policy-selected lock deadline. No built-in provisioning timeout replaces that policy.

Open-source tools

• pyca/cryptography
• rfc8785 (Python canonical JSON implementation)

Commercial / hosted tools

• AWS Key Management Service
• Azure Key Vault
• Google Cloud Key Management Service
• DigiCert Software Trust Manager
• GlobalSign Code Signing

Threat mappings

MITRE ATLAS

• AML.T0081 Modify AI Agent Configuration

MAESTRO

• Agent Goal Manipulation (L7)

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-5.2.1 Agent Profile Tampering
• AITech-5.2 Configuration Persistence

Google Secure AI Framework 2.0 - Risks

• N/A

Databricks AI Security Framework 3.0

• Agents - Core 13.6: Intent Breaking & Goal Manipulation