Testing Rule and Memory Inheritance Between AI Agents in FROST

What you will build

An AI agent in this prototype is a principal with an identifier, an optional parent, a private memory scope, and a declared capability set. FROST remains responsible for creating and coordinating agents. A separate governance component makes the final authorization decision before storage or tool execution.

The finished prototype demonstrates four properties:

1. A child can resolve a visible value from an ancestor when it has no local value.
2. A child can create and update its own memory but cannot change or delete ancestor-owned records.
3. A descendant cannot gain a capability that an ancestor did not pass down.
4. An invalid standard operating procedure (SOP) is rejected in full before its first step executes.

The tests inspect stored state, authorization decisions, and executed side effects. They do not treat an agent’s final message as proof that a control worked.

Concrete case: a report-production hierarchy

A root agent named report_manager owns project policy and delegates research to researcher. The researcher creates a narrower child named formatter.

The manager owns these records:

• project.client_region = "EU", visible to all descendants;
• policy.external_upload = false, visible to all descendants;
• report.retention_days = 30, visible to all descendants but writable only by its owner;
• credentials.storage_token, private to the manager.

The researcher may read the region and retention period, write its own notes, summarize documents, and save an internal artifact. The hostile scenario asks it to change the manager’s retention setting and run an SOP that uploads source documents to an external service.

A rule written only in a system prompt tests model compliance, not access control. The stronger invariant is that a forbidden operation never reaches storage or a tool, regardless of the model’s explanation or intent.

Define the invariants before writing code

Use explicit invariants so that framework behavior, model behavior, and security behavior do not become mixed together:

1. Provenance is preserved. A resolved value always retains its original owner.
2. Visibility does not imply mutability. Reading an ancestor record never creates permission to update it.
3. Capabilities narrow down the hierarchy. A child’s effective set is never wider than its parent’s effective set.
4. Authorization precedes execution. Every SOP step is validated before any handler runs.
5. Unknown actions fail closed. Adding an unfamiliar step type does not create an implicit permission.
6. Denied operations leave evidence. The decision log records the subject, operation, resource, policy version, and reason.

These invariants are independent of language-model quality. A stronger model may propose better plans, but it receives no special route around the policy boundary.

Model memory as records, not a merged dictionary

Agent memory cannot safely be represented as one recursively merged dictionary. Merging removes provenance, and without provenance the system cannot prove who owns a value or who may modify it.

Each stored record needs at least:

{
  "namespace": "report",
  "key": "retention_days",
  "value": 30,
  "owner_agent_id": "report_manager",
  "visibility": "descendants",
  "mutable_by": "owner",
  "revision": 1
}

Reading and writing deliberately use different addressing rules:

• A read walks from the current agent toward the root and returns the nearest visible record.
• A write names an exact owner scope and is permitted only when the actor is that owner.

This distinction prevents a common bug: a child resolves a parent value and then accidentally writes the updated value back into the same parent record.

Effective capabilities are calculated as an intersection:

effective(child) =
    environment_capabilities
    ∩ declared(root)
    ∩ declared(intermediate_parent)
    ∩ declared(child)

This applies the principle of least privilege. Delegating a report does not automatically delegate every tool available to the manager.

Threat model

The prototype does not defend against a host administrator who can directly edit the process, policy code, or memory database. That requires process isolation, protected policy deployment, and a separately trusted persistence layer.

Step 1: create the project

You need Python 3.11 or newer and pytest. The policy core requires no model API key and no external service.

mkdir frost-governance-lab
cd frost-governance-lab

python -m venv .venv
source .venv/bin/activate
python -m pip install pytest

mkdir -p frost_governance tests
touch frost_governance/__init__.py
touch frost_governance/core.py
touch tests/test_governance.py

On Windows PowerShell, activate the environment with:

.\.venv\Scripts\Activate.ps1

The resulting layout is:

frost-governance-lab/
├── frost_governance/
│   ├── __init__.py
│   └── core.py
└── tests/
    └── test_governance.py

Step 2: implement the trusted governance layer

Save the following as frost_governance/core.py. This code does not imitate agent reasoning. It is an executable authorization boundary that belongs between FROST and real memory or tool handlers.

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Callable, Iterable, Literal


Visibility = Literal["private", "children", "descendants"]


class GovernanceError(RuntimeError):
    pass


class AccessDenied(GovernanceError):
    pass


class ValidationError(GovernanceError):
    pass


class RevisionConflict(GovernanceError):
    pass


@dataclass(frozen=True)
class Agent:
    agent_id: str
    parent_id: str | None
    declared_capabilities: frozenset[str]


@dataclass
class MemoryRecord:
    namespace: str
    key: str
    value: Any
    owner_agent_id: str
    visibility: Visibility = "private"
    revision: int = 1


@dataclass(frozen=True)
class SOPStep:
    step_id: str
    action: str
    arguments: dict[str, Any] = field(default_factory=dict)


@dataclass(frozen=True)
class SOP:
    sop_id: str
    version: int
    steps: tuple[SOPStep, ...]


@dataclass(frozen=True)
class Decision:
    allowed: bool
    subject: str
    operation: str
    resource: str
    policy_version: str
    reason: str


class Governance:
    KNOWN_ACTIONS = frozenset({
        "memory.read",
        "memory.write",
        "document.summarize",
        "artifact.save",
        "external.upload",
    })

    def __init__(
        self,
        agents: Iterable[Agent],
        environment_capabilities: Iterable[str],
        policy_version: str = "1",
    ) -> None:
        agent_list = list(agents)
        self.agents = {
            agent.agent_id: agent for agent in agent_list
        }
        if len(self.agents) != len(agent_list):
            raise ValidationError("duplicate agent_id")

        self.environment_capabilities = frozenset(
            environment_capabilities
        )
        self.policy_version = policy_version
        self.memory: dict[
            tuple[str, str, str], MemoryRecord
        ] = {}
        self.audit: list[Decision] = []
        self._validate_hierarchy()

    def _validate_hierarchy(self) -> None:
        for agent in self.agents.values():
            if (
                agent.parent_id is not None
                and agent.parent_id not in self.agents
            ):
                raise ValidationError(
                    f"unknown parent: {agent.parent_id}"
                )

            seen: set[str] = set()
            cursor: Agent | None = agent
            while cursor is not None:
                if cursor.agent_id in seen:
                    raise ValidationError(
                        "agent hierarchy contains a cycle"
                    )
                seen.add(cursor.agent_id)
                cursor = (
                    self.agents[cursor.parent_id]
                    if cursor.parent_id is not None
                    else None
                )

    def lineage(self, agent_id: str) -> list[str]:
        if agent_id not in self.agents:
            raise ValidationError(f"unknown agent: {agent_id}")

        result: list[str] = []
        cursor = self.agents[agent_id]
        while True:
            result.append(cursor.agent_id)
            if cursor.parent_id is None:
                return result
            cursor = self.agents[cursor.parent_id]

    def is_descendant(
        self,
        candidate_id: str,
        ancestor_id: str,
    ) -> bool:
        return ancestor_id in self.lineage(candidate_id)[1:]

    def is_direct_child(
        self,
        candidate_id: str,
        parent_id: str,
    ) -> bool:
        return self.agents[candidate_id].parent_id == parent_id

    def effective_capabilities(
        self,
        agent_id: str,
    ) -> frozenset[str]:
        effective = set(self.environment_capabilities)

        for item_id in reversed(self.lineage(agent_id)):
            effective.intersection_update(
                self.agents[item_id].declared_capabilities
            )

        return frozenset(effective)

    def _decision(
        self,
        *,
        allowed: bool,
        subject: str,
        operation: str,
        resource: str,
        reason: str,
    ) -> Decision:
        decision = Decision(
            allowed=allowed,
            subject=subject,
            operation=operation,
            resource=resource,
            policy_version=self.policy_version,
            reason=reason,
        )
        self.audit.append(decision)
        return decision

    def put_initial(self, record: MemoryRecord) -> None:
        if record.owner_agent_id not in self.agents:
            raise ValidationError("unknown memory owner")

        storage_key = (
            record.owner_agent_id,
            record.namespace,
            record.key,
        )
        if storage_key in self.memory:
            raise ValidationError("initial record already exists")

        self.memory[storage_key] = record

    def _visible(
        self,
        reader_id: str,
        record: MemoryRecord,
    ) -> bool:
        if reader_id == record.owner_agent_id:
            return True
        if record.visibility == "private":
            return False
        if record.visibility == "children":
            return self.is_direct_child(
                reader_id,
                record.owner_agent_id,
            )
        return self.is_descendant(
            reader_id,
            record.owner_agent_id,
        )

    def resolve(
        self,
        reader_id: str,
        namespace: str,
        key: str,
    ) -> MemoryRecord:
        resource = f"{namespace}.{key}"

        if "memory.read" not in self.effective_capabilities(
            reader_id
        ):
            self._decision(
                allowed=False,
                subject=reader_id,
                operation="memory.read",
                resource=resource,
                reason="capability_not_inherited",
            )
            raise AccessDenied("memory.read is not allowed")

        for owner_id in self.lineage(reader_id):
            record = self.memory.get(
                (owner_id, namespace, key)
            )
            if record is None:
                continue

            if self._visible(reader_id, record):
                self._decision(
                    allowed=True,
                    subject=reader_id,
                    operation="memory.read",
                    resource=resource,
                    reason=f"visible_from:{owner_id}",
                )
                return record

            self._decision(
                allowed=False,
                subject=reader_id,
                operation="memory.read",
                resource=resource,
                reason=f"not_visible_from:{owner_id}",
            )
            raise AccessDenied("record is not visible")

        self._decision(
            allowed=False,
            subject=reader_id,
            operation="memory.read",
            resource=resource,
            reason="record_not_found",
        )
        raise KeyError(resource)

    def write(
        self,
        *,
        actor_id: str,
        owner_id: str,
        namespace: str,
        key: str,
        value: Any,
        expected_revision: int | None = None,
        visibility: Visibility = "private",
    ) -> MemoryRecord:
        resource = f"{owner_id}:{namespace}.{key}"

        if "memory.write" not in self.effective_capabilities(
            actor_id
        ):
            self._decision(
                allowed=False,
                subject=actor_id,
                operation="memory.write",
                resource=resource,
                reason="capability_not_inherited",
            )
            raise AccessDenied("memory.write is not allowed")

        if actor_id != owner_id:
            self._decision(
                allowed=False,
                subject=actor_id,
                operation="memory.write",
                resource=resource,
                reason="actor_is_not_owner",
            )
            raise AccessDenied(
                "cannot modify another agent's memory"
            )

        storage_key = (owner_id, namespace, key)
        current = self.memory.get(storage_key)

        if current is not None:
            if (
                expected_revision is not None
                and current.revision != expected_revision
            ):
                self._decision(
                    allowed=False,
                    subject=actor_id,
                    operation="memory.write",
                    resource=resource,
                    reason="revision_conflict",
                )
                raise RevisionConflict(resource)

            current.value = value
            current.visibility = visibility
            current.revision += 1
            result = current
        else:
            result = MemoryRecord(
                namespace=namespace,
                key=key,
                value=value,
                owner_agent_id=owner_id,
                visibility=visibility,
            )
            self.memory[storage_key] = result

        self._decision(
            allowed=True,
            subject=actor_id,
            operation="memory.write",
            resource=resource,
            reason="owner_write",
        )
        return result

    def validate_sop(
        self,
        actor_id: str,
        sop: SOP,
    ) -> tuple[Decision, ...]:
        if sop.version < 1:
            raise ValidationError("SOP version must be positive")
        if not sop.steps:
            raise ValidationError(
                "SOP must contain at least one step"
            )
        if len(sop.steps) > 50:
            raise ValidationError("SOP is too large")

        capabilities = self.effective_capabilities(actor_id)
        decisions: list[Decision] = []
        seen_ids: set[str] = set()

        for step in sop.steps:
            resource = f"{sop.sop_id}:{step.step_id}"

            if not step.step_id:
                reason = "empty_step_id"
            elif step.step_id in seen_ids:
                reason = "duplicate_step_id"
            elif step.action not in self.KNOWN_ACTIONS:
                reason = "unknown_action"
            elif step.action not in capabilities:
                reason = "action_not_in_effective_capabilities"
            elif (
                step.action == "memory.write"
                and step.arguments.get("owner_agent_id")
                != actor_id
            ):
                reason = "cross_owner_memory_write"
            else:
                reason = "step_allowed"

            allowed = reason == "step_allowed"
            decision = Decision(
                allowed=allowed,
                subject=actor_id,
                operation="sop.validate",
                resource=resource,
                policy_version=self.policy_version,
                reason=reason,
            )
            decisions.append(decision)
            seen_ids.add(step.step_id)

        self.audit.extend(decisions)

        rejected = [
            item for item in decisions if not item.allowed
        ]
        if rejected:
            reasons = ", ".join(
                item.reason for item in rejected
            )
            raise AccessDenied(f"SOP rejected: {reasons}")

        return tuple(decisions)

    def execute_sop(
        self,
        actor_id: str,
        sop: SOP,
        handlers: dict[
            str, Callable[[dict[str, Any]], Any]
        ],
    ) -> list[Any]:
        self.validate_sop(actor_id, sop)

        missing_handlers = {
            step.action
            for step in sop.steps
            if step.action not in handlers
        }
        if missing_handlers:
            missing = ", ".join(sorted(missing_handlers))
            raise ValidationError(
                f"missing handlers: {missing}"
            )

        results: list[Any] = []
        for step in sop.steps:
            results.append(
                handlers[step.action](step.arguments)
            )
        return results

validate_sop evaluates every step before dispatch. If the final step is forbidden, the system does not run the preceding allowed steps. KNOWN_ACTIONS is an allowlist: an unfamiliar action is denied even if its name looks harmless.

Handler availability is checked after authorization but still before execution. This avoids starting a process that cannot complete because one local adapter is missing.

Step 3: add reproducible tests

Save the following as tests/test_governance.py. The placeholder token is synthetic test data; do not insert a real secret.

import pytest

from frost_governance.core import (
    AccessDenied,
    Agent,
    Governance,
    MemoryRecord,
    RevisionConflict,
    SOP,
    SOPStep,
)


ALL = frozenset({
    "memory.read",
    "memory.write",
    "document.summarize",
    "artifact.save",
    "external.upload",
})


@pytest.fixture
def governance() -> Governance:
    manager = Agent(
        agent_id="report_manager",
        parent_id=None,
        declared_capabilities=ALL,
    )
    researcher = Agent(
        agent_id="researcher",
        parent_id="report_manager",
        declared_capabilities=frozenset({
            "memory.read",
            "memory.write",
            "document.summarize",
            "artifact.save",
        }),
    )
    formatter = Agent(
        agent_id="formatter",
        parent_id="researcher",
        declared_capabilities=frozenset({
            "memory.read",
            "artifact.save",
        }),
    )

    service = Governance(
        agents=[manager, researcher, formatter],
        environment_capabilities=ALL,
        policy_version="lab-1",
    )

    service.put_initial(MemoryRecord(
        namespace="project",
        key="client_region",
        value="EU",
        owner_agent_id="report_manager",
        visibility="descendants",
    ))
    service.put_initial(MemoryRecord(
        namespace="policy",
        key="external_upload",
        value=False,
        owner_agent_id="report_manager",
        visibility="descendants",
    ))
    service.put_initial(MemoryRecord(
        namespace="report",
        key="retention_days",
        value=30,
        owner_agent_id="report_manager",
        visibility="descendants",
    ))
    service.put_initial(MemoryRecord(
        namespace="credentials",
        key="storage_token",
        value="test-only-placeholder",
        owner_agent_id="report_manager",
        visibility="private",
    ))
    return service


def test_child_reads_visible_parent_memory(governance):
    record = governance.resolve(
        "researcher",
        "project",
        "client_region",
    )

    assert record.value == "EU"
    assert record.owner_agent_id == "report_manager"
    assert governance.audit[-1].allowed is True
    assert governance.audit[-1].reason == (
        "visible_from:report_manager"
    )


def test_grandchild_reads_descendant_memory(governance):
    record = governance.resolve(
        "formatter",
        "report",
        "retention_days",
    )

    assert record.value == 30
    assert record.owner_agent_id == "report_manager"


def test_private_parent_memory_is_hidden(governance):
    with pytest.raises(AccessDenied):
        governance.resolve(
            "researcher",
            "credentials",
            "storage_token",
        )

    assert governance.audit[-1].allowed is False
    assert governance.audit[-1].reason == (
        "not_visible_from:report_manager"
    )


def test_child_can_write_its_own_memory(governance):
    created = governance.write(
        actor_id="researcher",
        owner_id="researcher",
        namespace="notes",
        key="summary",
        value="Draft is ready",
        visibility="children",
    )

    assert created.owner_agent_id == "researcher"
    assert created.revision == 1
    assert governance.resolve(
        "formatter", "notes", "summary"
    ).value == "Draft is ready"


def test_child_cannot_change_parent_memory(governance):
    before = governance.resolve(
        "report_manager",
        "report",
        "retention_days",
    )
    original_revision = before.revision

    with pytest.raises(AccessDenied):
        governance.write(
            actor_id="researcher",
            owner_id="report_manager",
            namespace="report",
            key="retention_days",
            value=0,
            expected_revision=original_revision,
        )

    after = governance.resolve(
        "report_manager",
        "report",
        "retention_days",
    )
    assert after.value == 30
    assert after.revision == original_revision

    denied = [
        item for item in governance.audit
        if item.reason == "actor_is_not_owner"
    ]
    assert len(denied) == 1


def test_revision_conflict_preserves_newer_value(governance):
    first = governance.write(
        actor_id="researcher",
        owner_id="researcher",
        namespace="notes",
        key="status",
        value="started",
    )

    governance.write(
        actor_id="researcher",
        owner_id="researcher",
        namespace="notes",
        key="status",
        value="reviewed",
        expected_revision=first.revision,
    )

    with pytest.raises(RevisionConflict):
        governance.write(
            actor_id="researcher",
            owner_id="researcher",
            namespace="notes",
            key="status",
            value="stale update",
            expected_revision=1,
        )

    current = governance.resolve(
        "researcher", "notes", "status"
    )
    assert current.value == "reviewed"
    assert current.revision == 2


def test_child_cannot_add_parent_capability(governance):
    researcher_caps = governance.effective_capabilities(
        "researcher"
    )
    formatter_caps = governance.effective_capabilities(
        "formatter"
    )

    assert "external.upload" not in researcher_caps
    assert "external.upload" not in formatter_caps
    assert "memory.write" not in formatter_caps


def test_valid_sop_executes_all_steps(governance):
    effects = []

    sop = SOP(
        sop_id="prepare-report",
        version=1,
        steps=(
            SOPStep(
                "summarize",
                "document.summarize",
                {"document_id": "fixture-1"},
            ),
            SOPStep(
                "save",
                "artifact.save",
                {"path": "internal/report.txt"},
            ),
        ),
    )

    handlers = {
        "document.summarize": (
            lambda arguments: effects.append(
                ("summarize", arguments["document_id"])
            )
        ),
        "artifact.save": (
            lambda arguments: effects.append(
                ("save", arguments["path"])
            )
        ),
    }

    governance.execute_sop(
        "researcher", sop, handlers
    )

    assert effects == [
        ("summarize", "fixture-1"),
        ("save", "internal/report.txt"),
    ]


def test_forbidden_sop_has_zero_side_effects(governance):
    effects = []

    sop = SOP(
        sop_id="unsafe-report",
        version=1,
        steps=(
            SOPStep(
                "summarize",
                "document.summarize",
                {"document_id": "fixture-1"},
            ),
            SOPStep(
                "save",
                "artifact.save",
                {"path": "internal/report.txt"},
            ),
            SOPStep(
                "upload",
                "external.upload",
                {"destination": "untrusted-service"},
            ),
        ),
    )

    handlers = {
        "document.summarize": (
            lambda arguments: effects.append("summarize")
        ),
        "artifact.save": (
            lambda arguments: effects.append("save")
        ),
        "external.upload": (
            lambda arguments: effects.append("upload")
        ),
    }

    with pytest.raises(
        AccessDenied,
        match="action_not_in_effective_capabilities",
    ):
        governance.execute_sop(
            "researcher", sop, handlers
        )

    assert effects == []


def test_cross_owner_write_sop_is_rejected(governance):
    sop = SOP(
        sop_id="rewrite-manager-memory",
        version=1,
        steps=(
            SOPStep(
                "change-retention",
                "memory.write",
                {
                    "owner_agent_id": "report_manager",
                    "namespace": "report",
                    "key": "retention_days",
                    "value": 0,
                },
            ),
        ),
    )

    with pytest.raises(
        AccessDenied,
        match="cross_owner_memory_write",
    ):
        governance.validate_sop("researcher", sop)


def test_unknown_action_is_denied(governance):
    sop = SOP(
        sop_id="unknown-action",
        version=1,
        steps=(
            SOPStep(
                "run-shell",
                "shell.eval",
                {"command": "not executed"},
            ),
        ),
    )

    with pytest.raises(
        AccessDenied,
        match="unknown_action",
    ):
        governance.validate_sop("researcher", sop)


def test_duplicate_step_ids_are_denied(governance):
    sop = SOP(
        sop_id="duplicate-steps",
        version=1,
        steps=(
            SOPStep(
                "same",
                "document.summarize",
                {},
            ),
            SOPStep(
                "same",
                "artifact.save",
                {},
            ),
        ),
    )

    with pytest.raises(
        AccessDenied,
        match="duplicate_step_id",
    ):
        governance.validate_sop("researcher", sop)


def test_audit_contains_policy_version(governance):
    governance.resolve(
        "researcher",
        "project",
        "client_region",
    )

    decision = governance.audit[-1]
    assert decision.subject == "researcher"
    assert decision.operation == "memory.read"
    assert decision.policy_version == "lab-1"

Step 4: run and interpret the suite

Execute:

python -m pytest -q

Do not copy an expected pass count into an operational report. Test discovery changes as the suite evolves. Instead, verify that the command exits with status 0 and that every discovered test is reported as passed.

For a more inspectable run:

python -m pytest -vv
python -m pytest -vv -k "parent_memory or forbidden_sop"
python -m pytest --collect-only -q

The critical evidence is:

• test_child_reads_visible_parent_memory returns a value whose owner remains report_manager;
• test_private_parent_memory_is_hidden proves that inheritance is selective;
• test_child_cannot_change_parent_memory proves both denial and unchanged storage state;
• test_child_cannot_add_parent_capability proves monotonic narrowing;
• test_forbidden_sop_has_zero_side_effects proves preflight rejection rather than late failure;
• the audit test proves that a decision is tied to a policy version.

Step 5: place the boundary around FROST

The framework adapter should never expose raw storage or unrestricted tool functions to an agent. Route every proposed operation through a narrow gateway:

class FrostGovernanceAdapter:
    def __init__(self, governance, tool_handlers):
        self.governance = governance
        self.tool_handlers = tool_handlers

    def read_memory(self, frost_agent_id, namespace, key):
        record = self.governance.resolve(
            frost_agent_id,
            namespace,
            key,
        )
        return {
            "value": record.value,
            "owner_agent_id": record.owner_agent_id,
            "revision": record.revision,
        }

    def write_memory(
        self,
        frost_agent_id,
        namespace,
        key,
        value,
        expected_revision=None,
    ):
        return self.governance.write(
            actor_id=frost_agent_id,
            owner_id=frost_agent_id,
            namespace=namespace,
            key=key,
            value=value,
            expected_revision=expected_revision,
        )

    def run_sop(self, frost_agent_id, sop):
        return self.governance.execute_sop(
            frost_agent_id,
            sop,
            self.tool_handlers,
        )

When FROST creates a child, register the child’s stable identifier, parent identifier, and requested capability set. Do not trust a parent identifier supplied later inside model-generated arguments. The runtime should bind identity to the execution context.

# Pseudocode: adapt these hooks to your FROST version.
child = frost.create_agent(
    name="researcher",
    parent=manager,
)

governance.registered_identity = {
    child.runtime_id: "researcher"
}

result = adapter.read_memory(
    frost_agent_id="researcher",
    namespace="project",
    key="client_region",
)

The important property is not the exact FROST method name. It is that the trusted adapter derives actor_id from the runtime session and never accepts an arbitrary actor identity from the generated SOP.

Validate serialized SOP input before authorization

Production agents will usually propose an SOP as JSON. Parsing valid JSON is not authorization. Validate its shape before constructing domain objects:

• require a non-empty sop_id and positive integer version;
• limit the number of steps and argument size;
• require unique, non-empty step identifiers;
• require action names to be strings;
• reject unexpected top-level fields when strict compatibility matters;
• reject references to agents or resources outside the current execution scope.

A minimal parser can look like this:

def parse_sop(payload: dict) -> SOP:
    if not isinstance(payload, dict):
        raise ValidationError("SOP must be an object")

    required = {"sop_id", "version", "steps"}
    if set(payload) != required:
        raise ValidationError("unexpected SOP fields")

    if not isinstance(payload["sop_id"], str):
        raise ValidationError("invalid sop_id")
    if not isinstance(payload["version"], int):
        raise ValidationError("invalid version")
    if not isinstance(payload["steps"], list):
        raise ValidationError("steps must be a list")
    if len(payload["steps"]) > 50:
        raise ValidationError("too many steps")

    steps = []
    for raw in payload["steps"]:
        if not isinstance(raw, dict):
            raise ValidationError("step must be an object")
        if set(raw) - {"step_id", "action", "arguments"}:
            raise ValidationError("unexpected step fields")
        if not isinstance(raw.get("step_id"), str):
            raise ValidationError("invalid step_id")
        if not isinstance(raw.get("action"), str):
            raise ValidationError("invalid action")

        arguments = raw.get("arguments", {})
        if not isinstance(arguments, dict):
            raise ValidationError("invalid arguments")

        steps.append(SOPStep(
            step_id=raw["step_id"],
            action=raw["action"],
            arguments=arguments,
        ))

    return SOP(
        sop_id=payload["sop_id"],
        version=payload["version"],
        steps=tuple(steps),
    )

Failure cases the happy path misses

Local shadowing hides a policy record

The current resolver returns the nearest visible value. That behavior is useful for ordinary preferences but dangerous for mandatory policy. A child could create policy.external_upload = true in its own scope and shadow the inherited value.

Separate memory classes in production:

• Preferences may use nearest-value shadowing.
• Mandatory policy must be combined monotonically or resolved from an authoritative policy store.
• Secrets should use explicit grants rather than hierarchical lookup.

For Boolean deny rules, a safe merge can require every applicable level to allow the operation. More complex policy needs typed combination logic, not generic dictionary merging.

The check and the mutation are separate

The in-memory prototype performs them in one process. With a database, checking revision 4 and later issuing an unconditional update creates a time-of-check/time-of-use race. Use an atomic statement equivalent to:

UPDATE memory
SET value = :value, revision = revision + 1
WHERE owner_agent_id = :owner
  AND namespace = :namespace
  AND key = :key
  AND revision = :expected_revision;

Treat zero updated rows as a revision conflict.

A tool performs effects during validation

Tool discovery, schema lookup, or argument normalization must not trigger the tool. Preflight should operate on static schemas and policy data. Actual handlers run only after the full SOP is accepted.

A child spoofs its identity

Never authorize using actor_id from model output. Bind the identity to the FROST task, worker token, or process context. Arguments may identify a target resource, but not the authenticated subject.

A known action accepts an unsafe destination

Action-level permission is necessary but insufficient. If artifact.save permits any path, it may write outside the workspace. Add argument policies for paths, hosts, MIME types, record identifiers, size, and data classification.

The audit log can be edited by the actor

An audit log is useful only if denied agents cannot rewrite it. Send decisions to append-only or separately controlled storage. Avoid recording secret values or full sensitive prompts.

Retries duplicate external effects

For real tools, add idempotency: assign a stable operation identifier, enforce uniqueness in the effect store, and return the recorded result on retry. Authorization and duplicate prevention solve different problems.

Extend the suite with adversarial cases

After the basic suite passes, add parameterized tests for:

• unknown parent identifiers and cyclic hierarchies;
• duplicate agent identifiers;
• direct-child visibility versus grandchild access;
• an agent without memory.read requesting its own record;
• a formatter attempting to grant itself memory.write;
• an empty SOP, oversized SOP, invalid version, and duplicate step IDs;
• a valid action with an invalid resource target;
• a handler missing after all policy checks pass;
• two writes using the same expected revision;
• a denied request containing a prompt injection that claims administrator approval.

The injection case should produce the same denial as a plain request. Policy code must evaluate structured identity, ownership, capabilities, and arguments—not persuasive natural language.

Production hardening checklist

• Keep policy enforcement outside the model process.
• Use stable runtime identities and validate the hierarchy when agents are created.
• Store provenance and revision with every memory record.
• Separate preferences, mandatory policy, credentials, and temporary task state.
• Compute effective capability sets server-side.
• Validate the complete SOP and every argument before dispatch.
• Execute through registered handlers rather than arbitrary function names.
• Use atomic conditional writes in persistent storage.
• Record allowed and denied decisions with a policy version.
• Redact values and credentials from decision logs.
• Add timeouts, quotas, output limits, and cancellation.
• Use operation identifiers for retry-safe external effects.
• Run the suite whenever FROST, the policy adapter, schemas, or agent templates change.

Limitations

This prototype is intentionally small. It does not provide:

• persistent or tamper-resistant storage;
• distributed transactions across several tools;
• cryptographic agent identity;
• revocation of already issued worker credentials;
• resource-level rules for every tool argument;
• rollback after a handler performs an irreversible effect;
• protection against a compromised host or policy administrator;
• formal verification of the policy engine.

Preflight validation provides atomic admission, not atomic execution. Once an accepted multi-step SOP begins, a later handler can still fail. Use compensating actions, transactional services, or a durable workflow engine when all-or-nothing execution is required.

Final acceptance criteria

The lab is complete when you can independently demonstrate all of the following:

1. The researcher reads project.client_region and the returned record still names report_manager as owner.
2. The formatter reads descendant-visible memory but cannot use memory.write.
3. The manager’s private token is not returned to either descendant.
4. The researcher writes its own note successfully.
5. An attempted write to manager-owned retention data raises a denial and leaves both value and revision unchanged.
6. The unsafe upload SOP produces zero handler effects.
7. An unknown action fails closed.
8. Every relevant decision contains the subject, resource, reason, and policy version.

If any criterion is verified only by the agent saying that it complied, it is not yet verified. Repeat the check against storage state, handler effects, and the decision record.

The rule to keep

Inheritance should transfer context without transferring ownership. A child may observe selected ancestor state, but mutation remains scoped to the owner; capabilities can narrow but cannot expand; and a process is admitted only after every step is known and authorized.

That boundary turns hierarchy from a prompt convention into a testable system property. FROST can continue coordinating agents, while deterministic code decides what those agents are actually allowed to read, change, and execute.

← Browse all guides · Open the Agent Lab glossary →