Hands-On: Audit an Isolated Agent Testbed Before First Run
This is Hands-On 13A in the Local-First Agent Operations series. It accompanies Part 13, Running Multiple Agents Safely on One LAN, where I separated a useful second-agent testbed from the much stronger claim that two environments are actually isolated.
It is much cheaper to find a shared state directory in a JSON file than after two agents have written into it. The same is true for a port collision, a reused credential identity, or a synchronized Vault with two writers. This lab is a small preflight for those mistakes.
The wording of its successful result is deliberate: MANIFEST READY FOR ISOLATED TEST. That means the declaration is internally consistent enough to begin live testing. It does not mean the machine was inspected, the agents are isolated, or the design is secure.
This lab ends at declaration readiness. Everything below that line belongs to a separately reviewed live acceptance procedure.
1. Start With What This Lab Refuses to Do
The package reads local JSON fixtures, validates a closed set of fields, and prints a deterministic report. It does not create an account, change a permission, inspect launchd, open a socket, connect over SSH, query a keychain, start an agent, invoke LLM-Ops-Kit, or write into a knowledge store.
That limitation is useful. A declaration preflight can run during design and code review without needing access to the target machine. It catches contradictions before they become an installation problem, then stops before the work requires privilege or private evidence.
The package is deliberately small:
agent-isolation-preflight/
|-- README.md
|-- manifest.json
|-- isolation_preflight.py
|-- run_lab.py
|-- test_isolation_preflight.py
`-- fixtures/
|-- ready.json
|-- shared-root.json
|-- shared-port.json
`-- shared-vault-writers.json
It requires Python 3.10 or newer and uses only the standard library.
Download the complete Hands-On 13A package and its SHA-256 checksum. Every source file is also available here through the site’s collapsed source viewer:
README.md markdown View source
# Agent Isolation Manifest Preflight
This model-free teaching package checks an invented two-agent declaration for obvious ownership collisions before anyone provisions or starts the environment. It uses Python 3.10 or newer and the standard library.
It does not inspect users, processes, launchd, files, ports, credentials, channels, synchronization, SSH, or a model server. `MANIFEST READY FOR ISOLATED TEST` means only that the declaration is internally consistent enough to begin a separately reviewed live test.
Run the ready manifest:
```bash
python3 -B isolation_preflight.py fixtures/ready.json
```
Exercise the collision fixtures:
```bash
python3 -B isolation_preflight.py fixtures/shared-root.json || test $? -eq 1
python3 -B isolation_preflight.py fixtures/shared-port.json || test $? -eq 1
python3 -B isolation_preflight.py fixtures/shared-vault-writers.json || test $? -eq 1
```
Run the fixed walkthrough and unit tests:
```bash
python3 -B run_lab.py
python3 -B -m unittest -v test_isolation_preflight.py
```
The ready fixture declares two different execution users, process-manager domains, root sets, credential canaries, channel canaries, and same-host listener ports. Its shared writable test Vault has one owner and one writer. An immutable shared corpus has no writer. Both clients declare attributable use of one remote inference dependency plus planned contention and failure canaries.
Collection fields are set-like and reject duplicate entries. Shared stores require at least two participating agents, non-shared stores allow no more than one participant, and the contention and endpoint-failure canaries must have different identifiers. These are declaration checks only; none of the listed tests is executed by this package.
The manifest contains symbolic names, not deployable paths or secrets. Replace none of those examples with private data in a publication artifact. A real collector and acceptance procedure belong behind a separate privilege, privacy, and approval boundary.
manifest.json json View source
{
"schema_version": 1,
"name": "agent-isolation-preflight",
"runtime": "python>=3.10",
"capabilities": ["read_supplied_manifest", "validate_declaration", "render_report"],
"forbidden_capabilities": ["provision_user", "inspect_live_system", "open_network", "start_process", "read_secret", "write_authoritative_store", "certify_security"],
"files": {
"README.md": "c76fdfe3d5609b5f1f6810e1d8d1926afe4c7f766b75e9b13d3e1d8a997e7614",
"fixtures/ready.json": "65c772d3f7418d624cb8e9e6d57db945d35980775ff67ef775e955b2b69786e7",
"fixtures/shared-port.json": "a288e47e3c91eaaa77ee96a5530139f71773e6d23faa07e64070566ceb621c8d",
"fixtures/shared-root.json": "7f21d10f9269adb54a6c0c1fa7382d108a06f20a98fca547a0df2631b52d8b42",
"fixtures/shared-vault-writers.json": "af0e9ce8c5aec7f23b21a1d3882cfc3109eae8995654a6c1844d8d02d157d051",
"isolation_preflight.py": "df6a269973b93cb4c876f8f711dcd36527bc9dba341cc93e5c623c30a39d215c",
"run_lab.py": "190e9bb1ee4ab0358ec3b052ba7b200873f6f8dbdad54433b60e849588b655f0",
"test_isolation_preflight.py": "26f40254bee1ee5a304326fba0b693be441585a85c32b612ab7f9392acbc85c0"
}
}
isolation_preflight.py python View source
#!/usr/bin/env python3
"""Validate an invented multi-agent isolation manifest without live inspection."""
from __future__ import annotations
import argparse
import json
import re
import sys
from pathlib import Path
from typing import Any
SYMBOL = re.compile(r"^[a-z][a-z0-9-]{0,47}$")
ROOT = re.compile(r"^[a-z][a-z0-9-]{0,31}:[a-z][a-z0-9-]{0,47}$")
FORBIDDEN_TEXT = (
re.compile(r"(?:^|[\s=:])(?:/|~/)[^\s]*"),
re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b"),
re.compile(r"\b[a-z][a-z0-9+.-]*://", re.IGNORECASE),
re.compile(r"\b(?:token|password|secret|api[_-]?key)\s*[:=]", re.IGNORECASE),
re.compile(r"-----BEGIN [A-Z ]+PRIVATE KEY-----"),
)
TOP_FIELDS = {"schema_version", "testbed_id", "agents", "authoritative_stores", "shared_inference"}
AGENT_FIELDS = {
"id",
"host",
"execution_user",
"process_manager",
"roots",
"credential_refs",
"channel_refs",
"listeners",
"teardown",
}
PROCESS_FIELDS = {"mode", "domain", "bootstrap_test_plan"}
ROOT_FIELDS = {"config", "state", "logs", "cache", "workspace", "memory"}
LISTENER_FIELDS = {"service", "port"}
TEARDOWN_FIELDS = {"remove", "preserve"}
STORE_FIELDS = {"id", "shared", "writable", "owner", "writers", "readers", "transport"}
INFERENCE_FIELDS = {"id", "clients", "request_identity", "contention_test", "failure_test"}
BOOTSTRAP_CHECKS = {"login", "logout", "reboot", "headless"}
class ManifestError(Exception):
"""A bounded structural error which never echoes rejected input."""
def __init__(self, code: str):
super().__init__(code)
self.code = code
def _object(value: Any, fields: set[str], code: str) -> dict[str, Any]:
if not isinstance(value, dict) or set(value) != fields:
raise ManifestError(code)
return value
def _array(value: Any, code: str, *, minimum: int = 0) -> list[Any]:
if not isinstance(value, list) or len(value) < minimum:
raise ManifestError(code)
return value
def _symbol(value: Any, code: str) -> str:
if not isinstance(value, str) or not SYMBOL.fullmatch(value):
raise ManifestError(code)
return value
def _root(value: Any, code: str) -> str:
if not isinstance(value, str) or not ROOT.fullmatch(value):
raise ManifestError(code)
return value
def _bool(value: Any, code: str) -> bool:
if type(value) is not bool:
raise ManifestError(code)
return value
def _privacy_scan(value: Any) -> None:
if isinstance(value, dict):
for item in value.values():
_privacy_scan(item)
elif isinstance(value, list):
for item in value:
_privacy_scan(item)
elif isinstance(value, str) and any(pattern.search(value) for pattern in FORBIDDEN_TEXT):
raise ManifestError("E_PRIVATE_TEXT")
def _unique(values: list[str]) -> bool:
return len(values) == len(set(values))
def _require_unique(values: list[str], code: str) -> None:
if not _unique(values):
raise ManifestError(code)
def validate_manifest(raw: Any) -> dict[str, Any]:
manifest = _object(raw, TOP_FIELDS, "E_TOP_LEVEL")
if manifest["schema_version"] != 1 or type(manifest["schema_version"]) is not int:
raise ManifestError("E_SCHEMA_VERSION")
_symbol(manifest["testbed_id"], "E_TESTBED_ID")
_privacy_scan(manifest)
agents = _array(manifest["agents"], "E_AGENTS", minimum=2)
agent_ids: list[str] = []
users: list[str] = []
domains: list[str] = []
roots: list[str] = []
credential_refs: list[str] = []
channel_refs: list[str] = []
listeners: list[tuple[str, int]] = []
for index, value in enumerate(agents):
agent = _object(value, AGENT_FIELDS, "E_AGENT_FIELDS")
agent_id = _symbol(agent["id"], "E_AGENT_ID")
host = _symbol(agent["host"], "E_HOST")
user = _symbol(agent["execution_user"], "E_EXECUTION_USER")
agent_ids.append(agent_id)
users.append(f"{host}:{user}")
process = _object(agent["process_manager"], PROCESS_FIELDS, "E_PROCESS_MANAGER")
mode = process["mode"]
if mode not in {"launchagent", "standalone"}:
raise ManifestError("E_PROCESS_MODE")
domain = _symbol(process["domain"], "E_PROCESS_DOMAIN")
domains.append(f"{host}:{domain}")
checks = [_symbol(item, "E_BOOTSTRAP_CHECK") for item in _array(process["bootstrap_test_plan"], "E_BOOTSTRAP_PLAN")]
_require_unique(checks, "E_BOOTSTRAP_DUPLICATE")
if mode == "launchagent" and set(checks) != BOOTSTRAP_CHECKS:
raise ManifestError("E_BOOTSTRAP_PLAN")
if mode == "standalone" and checks:
raise ManifestError("E_BOOTSTRAP_PLAN")
root_map = _object(agent["roots"], ROOT_FIELDS, "E_ROOT_FIELDS")
agent_roots = [_root(root_map[name], "E_ROOT_VALUE") for name in sorted(ROOT_FIELDS)]
if not _unique(agent_roots):
raise ManifestError("E_ROOT_COLLISION_LOCAL")
roots.extend(agent_roots)
credentials = [_symbol(item, "E_CREDENTIAL_REF") for item in _array(agent["credential_refs"], "E_CREDENTIAL_REFS", minimum=1)]
channels = [_symbol(item, "E_CHANNEL_REF") for item in _array(agent["channel_refs"], "E_CHANNEL_REFS", minimum=1)]
if not _unique(credentials) or not _unique(channels):
raise ManifestError("E_IDENTITY_REF_DUPLICATE")
credential_refs.extend(credentials)
channel_refs.extend(channels)
for item in _array(agent["listeners"], "E_LISTENERS", minimum=1):
listener = _object(item, LISTENER_FIELDS, "E_LISTENER_FIELDS")
_symbol(listener["service"], "E_SERVICE")
port = listener["port"]
if type(port) is not int or not 1024 <= port <= 65535:
raise ManifestError("E_PORT")
listeners.append((host, port))
teardown = _object(agent["teardown"], TEARDOWN_FIELDS, "E_TEARDOWN_FIELDS")
remove = [_symbol(item, "E_TEARDOWN_REMOVE") for item in _array(teardown["remove"], "E_TEARDOWN_REMOVE", minimum=1)]
preserve = [_symbol(item, "E_TEARDOWN_PRESERVE") for item in _array(teardown["preserve"], "E_TEARDOWN_PRESERVE", minimum=1)]
_require_unique(remove, "E_TEARDOWN_REMOVE_DUPLICATE")
_require_unique(preserve, "E_TEARDOWN_PRESERVE_DUPLICATE")
if set(remove) & set(preserve):
raise ManifestError("E_TEARDOWN_CONFLICT")
findings: list[str] = []
if not _unique(agent_ids):
findings.append("duplicate_agent_identity")
if not _unique(users):
findings.append("shared_execution_user")
if not _unique(domains):
findings.append("shared_process_domain")
if not _unique(roots):
findings.append("shared_owned_root")
if not _unique(credential_refs):
findings.append("shared_credential_reference")
if not _unique(channel_refs):
findings.append("shared_channel_reference")
if not _unique([f"{host}:{port}" for host, port in listeners]):
findings.append("same_host_port_collision")
stores = _array(manifest["authoritative_stores"], "E_STORES", minimum=1)
store_ids: list[str] = []
known_agents = set(agent_ids)
for value in stores:
store = _object(value, STORE_FIELDS, "E_STORE_FIELDS")
store_id = _symbol(store["id"], "E_STORE_ID")
store_ids.append(store_id)
shared = _bool(store["shared"], "E_STORE_SHARED")
writable = _bool(store["writable"], "E_STORE_WRITABLE")
owner = store["owner"]
if owner is not None:
owner = _symbol(owner, "E_STORE_OWNER")
writers = [_symbol(item, "E_STORE_WRITER") for item in _array(store["writers"], "E_STORE_WRITERS")]
readers = [_symbol(item, "E_STORE_READER") for item in _array(store["readers"], "E_STORE_READERS")]
_require_unique(writers, "E_STORE_WRITER_DUPLICATE")
_require_unique(readers, "E_STORE_READER_DUPLICATE")
transport = store["transport"]
if transport is not None:
_symbol(transport, "E_STORE_TRANSPORT")
if not set(writers + readers).issubset(known_agents):
raise ManifestError("E_STORE_AGENT")
if writable and (owner not in known_agents or writers != [owner]):
findings.append(f"store_owner_mismatch:{store_id}")
if not writable and (owner is not None or writers):
findings.append(f"readonly_store_has_writer:{store_id}")
if shared and len(writers) > 1:
findings.append(f"shared_store_multiple_writers:{store_id}")
if shared and transport is None:
findings.append(f"shared_store_transport_missing:{store_id}")
participants = ({owner} if owner is not None else set()) | set(writers) | set(readers)
if shared and len(participants) < 2:
findings.append(f"shared_store_insufficient_participants:{store_id}")
if not shared and len(participants) > 1:
findings.append(f"nonshared_store_multiple_participants:{store_id}")
if not _unique(store_ids):
findings.append("duplicate_store_identity")
inference = _object(manifest["shared_inference"], INFERENCE_FIELDS, "E_INFERENCE_FIELDS")
_symbol(inference["id"], "E_INFERENCE_ID")
clients = [_symbol(item, "E_INFERENCE_CLIENT") for item in _array(inference["clients"], "E_INFERENCE_CLIENTS", minimum=2)]
_require_unique(clients, "E_INFERENCE_CLIENT_DUPLICATE")
if set(clients) != known_agents:
findings.append("inference_client_set_incomplete")
if inference["request_identity"] != "per-client":
findings.append("inference_attribution_missing")
for name in ("contention_test", "failure_test"):
if not isinstance(inference[name], str) or not SYMBOL.fullmatch(inference[name]):
raise ManifestError("E_INFERENCE_TEST")
if inference["contention_test"] == inference["failure_test"]:
findings.append("inference_canaries_not_distinct")
return {
"schema_version": 1,
"testbed_id": manifest["testbed_id"],
"decision": "manifest_ready_for_isolated_test" if not findings else "not_ready",
"display_decision": "MANIFEST READY FOR ISOLATED TEST" if not findings else "NOT READY",
"agent_count": len(agents),
"shared_inference": inference["id"],
"findings": sorted(findings),
"runtime_observed": False,
"security_certified": False,
}
def render_text(result: dict[str, Any]) -> str:
lines = [
result["display_decision"],
f"Testbed: {result['testbed_id']}",
f"Declared agents: {result['agent_count']}",
f"Shared inference dependency: {result['shared_inference']}",
]
if result["findings"]:
lines.append("Findings:")
lines.extend(f"- {item}" for item in result["findings"])
else:
lines.append("Declaration conflicts: none")
lines.extend(
[
"Runtime observed: no",
"Security certified: no",
"Next gate: inspect the declared environment with separately reviewed tools.",
]
)
return "\n".join(lines) + "\n"
def load_manifest(path: Path) -> Any:
try:
return json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
raise ManifestError("E_INPUT") from exc
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("manifest", type=Path)
parser.add_argument("--format", choices=("text", "json"), default="text")
args = parser.parse_args(argv)
try:
result = validate_manifest(load_manifest(args.manifest))
except ManifestError as exc:
result = {"decision": "rejected", "error": exc.code, "runtime_observed": False, "security_certified": False}
print(json.dumps(result, sort_keys=True) if args.format == "json" else f"REJECTED\nError: {exc.code}\n")
return 2
print(json.dumps(result, indent=2, sort_keys=True) if args.format == "json" else render_text(result), end="" if args.format == "text" else "\n")
return 0 if result["decision"] == "manifest_ready_for_isolated_test" else 1
if __name__ == "__main__":
sys.exit(main())
run_lab.py python View source
#!/usr/bin/env python3
"""Run the fixed, model-free isolation preflight walkthrough."""
from __future__ import annotations
import json
from pathlib import Path
from isolation_preflight import validate_manifest
ROOT = Path(__file__).resolve().parent
CASES = {
"ready.json": ("manifest_ready_for_isolated_test", []),
"shared-root.json": ("not_ready", ["shared_owned_root"]),
"shared-port.json": ("not_ready", ["same_host_port_collision"]),
"shared-vault-writers.json": (
"not_ready",
["shared_store_multiple_writers:test-vault", "store_owner_mismatch:test-vault"],
),
}
def main() -> int:
passed = 0
for filename, (decision, required_findings) in CASES.items():
raw = json.loads((ROOT / "fixtures" / filename).read_text(encoding="utf-8"))
result = validate_manifest(raw)
checks = [
result["decision"] == decision,
all(item in result["findings"] for item in required_findings),
result["runtime_observed"] is False,
result["security_certified"] is False,
]
if not all(checks):
print(f"FAIL {filename}")
return 1
passed += len(checks)
print(f"PASS {filename}: {result['display_decision']}")
print(f"PASS {passed} of {len(CASES) * 4} conditions")
return 0
if __name__ == "__main__":
raise SystemExit(main())
test_isolation_preflight.py python View source
from __future__ import annotations
import ast
import copy
import hashlib
import json
import tempfile
import unittest
from contextlib import redirect_stdout
from io import StringIO
from pathlib import Path
import isolation_preflight as preflight
ROOT = Path(__file__).resolve().parent
def fixture(name: str = "ready.json") -> dict:
return json.loads((ROOT / "fixtures" / name).read_text(encoding="utf-8"))
class PreflightTests(unittest.TestCase):
def test_ready_manifest_is_declaration_ready(self) -> None:
result = preflight.validate_manifest(fixture())
self.assertEqual(result["display_decision"], "MANIFEST READY FOR ISOLATED TEST")
self.assertFalse(result["runtime_observed"])
self.assertFalse(result["security_certified"])
def test_ready_result_is_deterministic(self) -> None:
first = preflight.validate_manifest(fixture())
second = preflight.validate_manifest(fixture())
self.assertEqual(first, second)
self.assertEqual(preflight.render_text(first), preflight.render_text(second))
def test_shared_root_is_not_ready(self) -> None:
result = preflight.validate_manifest(fixture("shared-root.json"))
self.assertIn("shared_owned_root", result["findings"])
def test_same_host_port_collision_is_not_ready(self) -> None:
result = preflight.validate_manifest(fixture("shared-port.json"))
self.assertIn("same_host_port_collision", result["findings"])
def test_shared_store_rejects_multiple_writers(self) -> None:
result = preflight.validate_manifest(fixture("shared-vault-writers.json"))
self.assertIn("shared_store_multiple_writers:test-vault", result["findings"])
def test_store_rejects_duplicate_writer(self) -> None:
raw = fixture()
raw["authoritative_stores"][0]["writers"].append("agent-a")
with self.assertRaisesRegex(preflight.ManifestError, "E_STORE_WRITER_DUPLICATE"):
preflight.validate_manifest(raw)
def test_store_rejects_duplicate_reader(self) -> None:
raw = fixture()
raw["authoritative_stores"][0]["readers"].append("agent-a")
with self.assertRaisesRegex(preflight.ManifestError, "E_STORE_READER_DUPLICATE"):
preflight.validate_manifest(raw)
def test_readonly_store_may_have_zero_writers(self) -> None:
result = preflight.validate_manifest(fixture())
self.assertNotIn("readonly_store_has_writer:immutable-corpus", result["findings"])
def test_nonshared_store_with_one_participant_is_ready(self) -> None:
raw = fixture()
store = raw["authoritative_stores"][0]
store["shared"] = False
store["readers"] = ["agent-a"]
result = preflight.validate_manifest(raw)
self.assertEqual(result["decision"], "manifest_ready_for_isolated_test")
def test_nonshared_store_rejects_multiple_participants(self) -> None:
raw = fixture()
raw["authoritative_stores"][0]["shared"] = False
result = preflight.validate_manifest(raw)
self.assertIn("nonshared_store_multiple_participants:test-vault", result["findings"])
def test_shared_store_requires_two_participants(self) -> None:
raw = fixture()
store = raw["authoritative_stores"][0]
store["readers"] = ["agent-a"]
result = preflight.validate_manifest(raw)
self.assertIn("shared_store_insufficient_participants:test-vault", result["findings"])
def test_shared_store_with_two_participants_is_ready(self) -> None:
result = preflight.validate_manifest(fixture())
self.assertNotIn("shared_store_insufficient_participants:test-vault", result["findings"])
def test_writable_store_requires_one_declared_owner(self) -> None:
raw = fixture()
raw["authoritative_stores"][0]["owner"] = None
result = preflight.validate_manifest(raw)
self.assertIn("store_owner_mismatch:test-vault", result["findings"])
def test_launchagent_requires_complete_bootstrap_plan(self) -> None:
raw = fixture()
raw["agents"][0]["process_manager"]["bootstrap_test_plan"].remove("headless")
with self.assertRaisesRegex(preflight.ManifestError, "E_BOOTSTRAP_PLAN"):
preflight.validate_manifest(raw)
def test_launchagent_rejects_duplicate_bootstrap_check(self) -> None:
raw = fixture()
raw["agents"][0]["process_manager"]["bootstrap_test_plan"].append("login")
with self.assertRaisesRegex(preflight.ManifestError, "E_BOOTSTRAP_DUPLICATE"):
preflight.validate_manifest(raw)
def test_standalone_rejects_launchagent_bootstrap_plan(self) -> None:
raw = fixture()
raw["agents"][1]["process_manager"]["bootstrap_test_plan"] = ["login"]
with self.assertRaisesRegex(preflight.ManifestError, "E_BOOTSTRAP_PLAN"):
preflight.validate_manifest(raw)
def test_shared_credential_reference_is_not_ready(self) -> None:
raw = fixture()
raw["agents"][1]["credential_refs"] = list(raw["agents"][0]["credential_refs"])
result = preflight.validate_manifest(raw)
self.assertIn("shared_credential_reference", result["findings"])
def test_shared_channel_reference_is_not_ready(self) -> None:
raw = fixture()
raw["agents"][1]["channel_refs"] = list(raw["agents"][0]["channel_refs"])
result = preflight.validate_manifest(raw)
self.assertIn("shared_channel_reference", result["findings"])
def test_shared_inference_requires_all_clients(self) -> None:
raw = fixture()
raw["shared_inference"]["clients"] = ["agent-a", "agent-c"]
result = preflight.validate_manifest(raw)
self.assertIn("inference_client_set_incomplete", result["findings"])
def test_shared_inference_rejects_duplicate_client(self) -> None:
raw = fixture()
raw["shared_inference"]["clients"].append("agent-a")
with self.assertRaisesRegex(preflight.ManifestError, "E_INFERENCE_CLIENT_DUPLICATE"):
preflight.validate_manifest(raw)
def test_shared_inference_requires_attribution(self) -> None:
raw = fixture()
raw["shared_inference"]["request_identity"] = "shared"
result = preflight.validate_manifest(raw)
self.assertIn("inference_attribution_missing", result["findings"])
def test_shared_inference_requires_distinct_canaries(self) -> None:
raw = fixture()
raw["shared_inference"]["failure_test"] = raw["shared_inference"]["contention_test"]
result = preflight.validate_manifest(raw)
self.assertIn("inference_canaries_not_distinct", result["findings"])
def test_teardown_remove_and_preserve_must_not_overlap(self) -> None:
raw = fixture()
raw["agents"][0]["teardown"]["preserve"] = ["jobs-a"]
with self.assertRaisesRegex(preflight.ManifestError, "E_TEARDOWN_CONFLICT"):
preflight.validate_manifest(raw)
def test_teardown_rejects_duplicate_remove_item(self) -> None:
raw = fixture()
raw["agents"][0]["teardown"]["remove"].append("jobs-a")
with self.assertRaisesRegex(preflight.ManifestError, "E_TEARDOWN_REMOVE_DUPLICATE"):
preflight.validate_manifest(raw)
def test_teardown_rejects_duplicate_preserve_item(self) -> None:
raw = fixture()
raw["agents"][0]["teardown"]["preserve"].append("reviewed-notes-a")
with self.assertRaisesRegex(preflight.ManifestError, "E_TEARDOWN_PRESERVE_DUPLICATE"):
preflight.validate_manifest(raw)
def test_unknown_field_is_rejected(self) -> None:
raw = fixture()
raw["agents"][0]["unexpected"] = True
with self.assertRaisesRegex(preflight.ManifestError, "E_AGENT_FIELDS"):
preflight.validate_manifest(raw)
def test_boolean_port_is_rejected(self) -> None:
raw = fixture()
raw["agents"][0]["listeners"][0]["port"] = True
with self.assertRaisesRegex(preflight.ManifestError, "E_PORT"):
preflight.validate_manifest(raw)
def test_absolute_path_is_rejected_without_echo(self) -> None:
raw = fixture()
raw["testbed_id"] = "/private/example"
with self.assertRaises(preflight.ManifestError) as caught:
preflight.validate_manifest(raw)
self.assertNotIn("private", caught.exception.code)
def test_uri_is_rejected_without_echo(self) -> None:
raw = fixture()
raw["shared_inference"]["id"] = "ssh://private.invalid"
with self.assertRaises(preflight.ManifestError) as caught:
preflight.validate_manifest(raw)
self.assertNotIn("invalid", caught.exception.code)
def test_cli_rejection_is_bounded_json(self) -> None:
raw = fixture()
raw["schema_version"] = True
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "input.json"
path.write_text(json.dumps(raw), encoding="utf-8")
output = StringIO()
with redirect_stdout(output):
status = preflight.main([str(path), "--format", "json"])
payload = json.loads(output.getvalue())
self.assertEqual(status, 2)
self.assertEqual(payload["decision"], "rejected")
self.assertFalse(payload["runtime_observed"])
def test_runtime_source_has_no_execution_or_network_import(self) -> None:
source = (ROOT / "isolation_preflight.py").read_text(encoding="utf-8")
tree = ast.parse(source)
forbidden = {"subprocess", "socket", "http", "urllib", "requests", "paramiko", "llmops_kit"}
imported: set[str] = set()
for node in ast.walk(tree):
if isinstance(node, ast.Import):
imported.update(alias.name.split(".")[0] for alias in node.names)
elif isinstance(node, ast.ImportFrom) and node.module:
imported.add(node.module.split(".")[0])
self.assertFalse(imported & forbidden)
def test_manifest_hashes_match_listed_files(self) -> None:
manifest = json.loads((ROOT / "manifest.json").read_text(encoding="utf-8"))
for relative, expected in manifest["files"].items():
actual = hashlib.sha256((ROOT / relative).read_bytes()).hexdigest()
self.assertEqual(actual, expected, relative)
def test_validation_does_not_mutate_input(self) -> None:
raw = fixture()
before = copy.deepcopy(raw)
preflight.validate_manifest(raw)
self.assertEqual(raw, before)
if __name__ == "__main__":
unittest.main()
ready.json json View source
{
"schema_version": 1,
"testbed_id": "lan-agent-lab",
"agents": [
{
"id": "agent-a",
"host": "edge-node",
"execution_user": "operator-a",
"process_manager": {
"mode": "launchagent",
"domain": "user-domain-a",
"bootstrap_test_plan": ["login", "logout", "reboot", "headless"]
},
"roots": {
"config": "agent-a:config",
"state": "agent-a:state",
"logs": "agent-a:logs",
"cache": "agent-a:cache",
"workspace": "agent-a:workspace",
"memory": "agent-a:memory"
},
"credential_refs": ["credential-canary-a"],
"channel_refs": ["channel-canary-a"],
"listeners": [
{"service": "gateway", "port": 46101},
{"service": "dashboard", "port": 46102}
],
"teardown": {
"remove": ["jobs-a", "listeners-a", "runtime-a"],
"preserve": ["reviewed-notes-a"]
}
},
{
"id": "agent-b",
"host": "edge-node",
"execution_user": "operator-b",
"process_manager": {
"mode": "standalone",
"domain": "service-domain-b",
"bootstrap_test_plan": []
},
"roots": {
"config": "agent-b:config",
"state": "agent-b:state",
"logs": "agent-b:logs",
"cache": "agent-b:cache",
"workspace": "agent-b:workspace",
"memory": "agent-b:memory"
},
"credential_refs": ["credential-canary-b"],
"channel_refs": ["channel-canary-b"],
"listeners": [
{"service": "gateway", "port": 46201},
{"service": "dashboard", "port": 46202}
],
"teardown": {
"remove": ["jobs-b", "listeners-b", "runtime-b"],
"preserve": ["reviewed-notes-b"]
}
}
],
"authoritative_stores": [
{
"id": "test-vault",
"shared": true,
"writable": true,
"owner": "agent-a",
"writers": ["agent-a"],
"readers": ["agent-a", "agent-b"],
"transport": "sync-canary"
},
{
"id": "immutable-corpus",
"shared": true,
"writable": false,
"owner": null,
"writers": [],
"readers": ["agent-a", "agent-b"],
"transport": "readonly-copy"
}
],
"shared_inference": {
"id": "model-node",
"clients": ["agent-a", "agent-b"],
"request_identity": "per-client",
"contention_test": "queue-contention-canary",
"failure_test": "endpoint-failure-canary"
}
}
shared-root.json json View source
{
"schema_version": 1,
"testbed_id": "shared-root-lab",
"agents": [
{
"id": "agent-a", "host": "edge-node", "execution_user": "operator-a",
"process_manager": {"mode": "standalone", "domain": "service-domain-a", "bootstrap_test_plan": []},
"roots": {"config": "shared:config", "state": "agent-a:state", "logs": "agent-a:logs", "cache": "agent-a:cache", "workspace": "agent-a:workspace", "memory": "agent-a:memory"},
"credential_refs": ["credential-canary-a"], "channel_refs": ["channel-canary-a"],
"listeners": [{"service": "gateway", "port": 46101}],
"teardown": {"remove": ["runtime-a"], "preserve": ["reviewed-notes-a"]}
},
{
"id": "agent-b", "host": "edge-node", "execution_user": "operator-b",
"process_manager": {"mode": "standalone", "domain": "service-domain-b", "bootstrap_test_plan": []},
"roots": {"config": "shared:config", "state": "agent-b:state", "logs": "agent-b:logs", "cache": "agent-b:cache", "workspace": "agent-b:workspace", "memory": "agent-b:memory"},
"credential_refs": ["credential-canary-b"], "channel_refs": ["channel-canary-b"],
"listeners": [{"service": "gateway", "port": 46201}],
"teardown": {"remove": ["runtime-b"], "preserve": ["reviewed-notes-b"]}
}
],
"authoritative_stores": [{"id": "immutable-corpus", "shared": true, "writable": false, "owner": null, "writers": [], "readers": ["agent-a", "agent-b"], "transport": "readonly-copy"}],
"shared_inference": {"id": "model-node", "clients": ["agent-a", "agent-b"], "request_identity": "per-client", "contention_test": "queue-canary", "failure_test": "failure-canary"}
}
shared-port.json json View source
{
"schema_version": 1,
"testbed_id": "shared-port-lab",
"agents": [
{
"id": "agent-a", "host": "edge-node", "execution_user": "operator-a",
"process_manager": {"mode": "standalone", "domain": "service-domain-a", "bootstrap_test_plan": []},
"roots": {"config": "agent-a:config", "state": "agent-a:state", "logs": "agent-a:logs", "cache": "agent-a:cache", "workspace": "agent-a:workspace", "memory": "agent-a:memory"},
"credential_refs": ["credential-canary-a"], "channel_refs": ["channel-canary-a"],
"listeners": [{"service": "gateway", "port": 46101}],
"teardown": {"remove": ["runtime-a"], "preserve": ["reviewed-notes-a"]}
},
{
"id": "agent-b", "host": "edge-node", "execution_user": "operator-b",
"process_manager": {"mode": "standalone", "domain": "service-domain-b", "bootstrap_test_plan": []},
"roots": {"config": "agent-b:config", "state": "agent-b:state", "logs": "agent-b:logs", "cache": "agent-b:cache", "workspace": "agent-b:workspace", "memory": "agent-b:memory"},
"credential_refs": ["credential-canary-b"], "channel_refs": ["channel-canary-b"],
"listeners": [{"service": "gateway", "port": 46101}],
"teardown": {"remove": ["runtime-b"], "preserve": ["reviewed-notes-b"]}
}
],
"authoritative_stores": [{"id": "immutable-corpus", "shared": true, "writable": false, "owner": null, "writers": [], "readers": ["agent-a", "agent-b"], "transport": "readonly-copy"}],
"shared_inference": {"id": "model-node", "clients": ["agent-a", "agent-b"], "request_identity": "per-client", "contention_test": "queue-canary", "failure_test": "failure-canary"}
}
shared-vault-writers.json json View source
{
"schema_version": 1,
"testbed_id": "shared-writer-lab",
"agents": [
{
"id": "agent-a", "host": "edge-node", "execution_user": "operator-a",
"process_manager": {"mode": "standalone", "domain": "service-domain-a", "bootstrap_test_plan": []},
"roots": {"config": "agent-a:config", "state": "agent-a:state", "logs": "agent-a:logs", "cache": "agent-a:cache", "workspace": "agent-a:workspace", "memory": "agent-a:memory"},
"credential_refs": ["credential-canary-a"], "channel_refs": ["channel-canary-a"],
"listeners": [{"service": "gateway", "port": 46101}],
"teardown": {"remove": ["runtime-a"], "preserve": ["reviewed-notes-a"]}
},
{
"id": "agent-b", "host": "edge-node", "execution_user": "operator-b",
"process_manager": {"mode": "standalone", "domain": "service-domain-b", "bootstrap_test_plan": []},
"roots": {"config": "agent-b:config", "state": "agent-b:state", "logs": "agent-b:logs", "cache": "agent-b:cache", "workspace": "agent-b:workspace", "memory": "agent-b:memory"},
"credential_refs": ["credential-canary-b"], "channel_refs": ["channel-canary-b"],
"listeners": [{"service": "gateway", "port": 46201}],
"teardown": {"remove": ["runtime-b"], "preserve": ["reviewed-notes-b"]}
}
],
"authoritative_stores": [{"id": "test-vault", "shared": true, "writable": true, "owner": "agent-a", "writers": ["agent-a", "agent-b"], "readers": ["agent-a", "agent-b"], "transport": "sync-canary"}],
"shared_inference": {"id": "model-node", "clients": ["agent-a", "agent-b"], "request_identity": "per-client", "contention_test": "queue-canary", "failure_test": "failure-canary"}
}
2. Verify and Extract the Package
Download the reviewed archive and checksum into one directory, then verify the archive before opening it:
shasum -a 256 -c agent-isolation-preflight.zip.sha256
mkdir isolation-lab
cd isolation-lab
unzip ../agent-isolation-preflight.zip
cd agent-isolation-preflight
python3 --version
Nothing needs to be installed with pip. The package manifest declares only three capabilities: reading a supplied manifest, validating the declaration, and rendering a report. Provisioning, live inspection, networking, process startup, secret access, authoritative writes, and security certification are explicitly outside the package.
3. Read the Clean Declaration
Open fixtures/ready.json. It describes two invented agents sharing one modest host and one remote inference dependency. The examples use symbolic roots such as agent-a:state; they are labels for ownership comparison, not filesystem paths to paste into a real configuration.
The two agents declare different execution users, process-manager domains, configuration roots, state roots, log roots, caches, workspaces, memory stores, credential canaries, channel canaries, and listener ports. One uses a per-user LaunchAgent plan. Because that mode depends on a usable bootstrap domain, the manifest requires planned checks for login, logout, reboot, and headless operation. The other uses a symbolic standalone service domain and therefore does not claim those LaunchAgent checks.
The knowledge section demonstrates two different valid authorities. A writable shared test Vault has one declared owner and one writer. Both agents may read it, but only the owner may write. An immutable shared corpus has readers and no writer, which is correct for read-only input.
The remote inference section lists both clients, requires per-client request identity, and names separate contention and endpoint-failure canaries. Those names do not prove that the tests ran. They make the missing live work visible in the declaration.
4. Run the Ready Case
python3 -B isolation_preflight.py fixtures/ready.json
The report should begin with:
MANIFEST READY FOR ISOLATED TEST
Testbed: lan-agent-lab
Declared agents: 2
Shared inference dependency: model-node
Declaration conflicts: none
Runtime observed: no
Security certified: no
Next gate: inspect the declared environment with separately reviewed tools.
The last three lines are not boilerplate. They keep a clean design review from being mistaken for runtime evidence. JSON cannot tell us who owns a live process, whether a socket is already occupied, whether a credential canary is truly unavailable to the other account, or whether synchronization presented a complete file set to a reader.
For machine-readable output, add --format json:
python3 -B isolation_preflight.py fixtures/ready.json --format json
The JSON carries the same decision, agent count, shared dependency, findings, and two explicit false claims: runtime_observed and security_certified.
5. Give Both Agents the Same Root
The first broken fixture declares the same configuration root for both agents:
python3 -B isolation_preflight.py fixtures/shared-root.json
test $? -eq 1
It returns NOT READY and reports shared_owned_root. The preflight does not try to decide whether that sharing was convenient or intentional. These roots were declared as agent-owned, so sharing one contradicts the isolation plan.
This check covers configuration, state, logs, cache, workspace, and memory roots. It also rejects two categories inside one agent pointing to the same symbolic root. A real collector would later resolve symbolic ownership to actual paths and inspect permissions, links, mounts, and process behavior. This lab does none of that.
6. Make the Socket Collision Obvious
The next fixture gives both gateway declarations the same port on the same host:
python3 -B isolation_preflight.py fixtures/shared-port.json
test $? -eq 1
The finding is same_host_port_collision. The same port on different hosts could be valid, which is why the comparison uses the host and port together. The fixture does not include real addresses or private ports.
A live test still has more work. It must verify the bind address, process owner, transport direction, and teardown behavior. A stale tunnel or undeclared debug listener will not appear in a declaration no matter how carefully the JSON is validated.
7. Try Two Writers on One Shared Vault
Now run the knowledge-ownership failure:
python3 -B isolation_preflight.py fixtures/shared-vault-writers.json
test $? -eq 1
The fixture declares Agent A as owner but lists both agents as writers. The report identifies both the owner mismatch and multiple writers on a shared authority.
The rule is narrower than saying every store needs one writer. A read-only replica or immutable corpus may have none. Every writable authority does need one declared owner, and a shared authority may have at most one writer in this initial design. That does not turn synchronized storage into a transaction system. It simply prevents the test from beginning with an ownership contradiction.
8. Look at the LaunchAgent Gate
In the clean fixture, Agent A uses launchagent mode and carries this test plan:
["login", "logout", "reboot", "headless"]
Remove headless and run the preflight again. The manifest is rejected with E_BOOTSTRAP_PLAN because it no longer describes the complete acceptance shape required for that mode.
The program does not claim that any of those tests passed. A future live procedure has to observe the applicable bootstrap domain and the agent’s behavior under each condition. If the account cannot provide the required persistent context, the design can move to a separately reviewed standalone manager or another native service domain instead of pretending that a property list is enough.
9. Keep Credential Tests Harmless
The example uses credential-canary-a and credential-canary-b, not secret values and not production identities. The same pattern is used for communication channels. If the two agents declare the same reference, the preflight returns shared_credential_reference or shared_channel_reference.
In a live acceptance run, each canary should be least privilege and disposable. Agent A should be able to use its own canary, Agent B should be denied that canary, and the reverse should be tested separately. There is no reason to prove separation by attempting to read a real token or send a message through another agent’s production channel.
The validator also rejects recognizable private forms such as absolute paths, network addresses, URI authorities, credential assignments, and private-key markers. Rejection errors use bounded codes and do not echo the input. These checks are defense in depth for the teaching artifact, not a complete secret scanner.
10. Run the Fixed Walkthrough
The runner exercises the ready manifest and all three collision fixtures:
python3 -B run_lab.py
It currently checks sixteen conditions. Every case must return the expected decision and findings, and every result must say that runtime was not observed and security was not certified.
Then run the unit suite:
python3 -B -m unittest -v test_isolation_preflight.py
The thirty-three tests cover deterministic output, root and port collisions, writable and read-only knowledge rules, shared and non-shared participant cardinality, LaunchAgent bootstrap plans, duplicate collection entries, shared identity references, inference attribution, distinct contention and failure canaries, teardown conflicts, closed fields, strict JSON types, bounded privacy rejection, the command-line error contract, lack of execution or network imports, non-mutating validation, and manifest-file integrity.
An abstract-syntax-tree check verifies that the runtime source does not import subprocess, socket, HTTP, SSH, or LLM-Ops-Kit surfaces. That is a bounded claim about this reviewed package. It is not a general proof that Python programs cannot have side effects.
11. Turn the Declaration Into a Live Test Plan
Once the manifest is clean, the real work begins. A separately reviewed collector and procedure would need to compare each declared item with the machine:
| Declaration | Live evidence still required |
|---|---|
| Execution user and process domain | Observed process ownership and bootstrap behavior across the required conditions |
| Symbolic roots | Resolved paths, permissions, links, mounts, write tests, and negative cross-account tests |
| Listener ports | Owning process, bind address, transport direction, collision behavior, and verified removal |
| Credential and channel canaries | Positive use by the owner and denial to the other agent without exposing values |
| Knowledge ownership | One observed writer, read-only consumers, synchronization caveats, and no concurrent merge path |
| Shared inference | Per-client attribution plus controlled solo, contention, endpoint-failure, and recovery runs |
| Teardown declaration | Removed jobs, processes, sockets, canaries, and owned roots, with preserved data verified separately |
That live phase should bind its report to the exact artifact version and digest tested. It should also retain failures and unknown observations rather than quietly turning them into a pass.
12. Clean Up
Return to the directory above the disposable lab and remove it:
cd ../..
rm -rf isolation-lab
No service or operating-system state was changed by the package. The only files created during this walkthrough came from extracting the archive into the directory you just removed.
Current State
The companion is a model-free declaration preflight over four invented fixtures. The ready case has separate roots, identities, references, ports, writable ownership, read-only authority, shared-inference test declarations, and teardown responsibilities. Three negative cases expose the collisions most likely to make an early two-agent experiment misleading.
The package does not inspect or certify a live environment. Its successful result is permission to begin an isolated test, not evidence that the test passed.
Next Work
The next artifact would be a privacy-bounded live collector that resolves the symbolic declaration into process, filesystem, launchd, listener, canary, synchronization, and inference observations. It would require explicit privileges, sanitization, versioned evidence, cancellation and timeout behavior, and its own negative tests.
Only after that procedure and teardown have been exercised against the exact release artifact should the result graduate from MANIFEST READY FOR ISOLATED TEST to a bounded isolation acceptance claim.
Join the Discussion
Comments for this post live in GitHub Discussions. That keeps moderation in one place and gives the conversation a stable home.