Hands-On: Package a Read-Only Component Triage Skill
This is Hands-On 12A in the Local-First Agent Operations series. It accompanies Part 12, Packaging Agent Operations as Installable Skills, where I drew the ownership line between a skill and the control plane beneath it.
An operational skill ought to make a good procedure easier to repeat. It should not make it easier to slip past the operator. That is the line this lab is built to hold.
The package in this tutorial reads invented, already-sanitized evidence. It validates that evidence, compares a captured status snapshot with captured canonical plans, and produces a restart review packet. It does not connect to a live LLM-Ops-Kit installation, discover dependencies, execute a command, or change a service. Package the triage, not the orchestrator.
The lab can explain why a packet is ready, incomplete, stale, or rejected. It never crosses the execution boundary.
1. Start With the Boundary
There are three different claims hiding inside the phrase “read-only.” The requested operation may be observational. The selected subcommand may avoid changing component lifecycle. The complete invocation path may still have side effects.
That third claim matters here because the reviewed LLM-Ops-Kit entrypoint can apply an approved runtime update before dispatching an otherwise read-only command. I cannot prove a zero-mutation lab merely by calling something named status. The lab therefore starts with captured fixtures and never starts a subprocess.
The package contains a short skill contract, a closed schema, one deterministic renderer, three fixtures, a golden Markdown packet, an acceptance runner, and unit tests:
llmops-component-triage/
|-- SKILL.md
|-- README.md
|-- manifest.json
|-- schemas/evidence-envelope.schema.json
|-- scripts/triage_packet.py
|-- fixtures/complete.json
|-- fixtures/unobserved.json
|-- fixtures/stale.json
|-- fixtures/expected-complete.md
|-- run_lab.py
`-- test_triage_packet.py
The manifest declares four capabilities: reading a supplied fixture, validation, correlation, and rendering. Network, subprocess, secret access, discovery, and lifecycle execution are not capabilities of this package.
Download the complete Hands-On 12A 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
# LLM-Ops Component Triage Teaching Package
This model-free package turns invented, sanitized capture fixtures into deterministic component-restart review packets. It uses Python 3.10 or newer and the standard library. It does not invoke `llmops`, start a subprocess, use the network, discover a filesystem, retrieve a secret, or execute a lifecycle operation.
Run the complete example:
```bash
python3 -B scripts/triage_packet.py fixtures/complete.json --as-of 2030-01-02T03:09:05Z --format markdown
```
Compare the incomplete and stale cases:
```bash
python3 -B scripts/triage_packet.py fixtures/unobserved.json --as-of 2030-01-02T03:09:05Z --format markdown
python3 -B scripts/triage_packet.py fixtures/stale.json --as-of 2030-01-02T03:09:05Z --format json
```
Run acceptance and unit checks:
```bash
python3 -B run_lab.py
python3 -B -m unittest -v test_triage_packet.py
```
The fixtures are teaching records, not current operational evidence. A live collector and any executor are deliberately out of scope.
SKILL.md markdown View source
# Component Triage
Use this skill when an operator needs a review packet for a proposed cascading component restart and has supplied one already-sanitized evidence envelope conforming to `schemas/evidence-envelope.schema.json`.
Validate the envelope before interpretation. Treat the captured target-only and cascade plans as the canonical plan evidence. Correlate status only with component membership already present in that cascade plan; never infer or rediscover dependencies. Report missing, unreachable, unobserved, inconsistent, or stale evidence as not ready.
Show the equivalent operation only as a validated argument array. Never construct shell text, execute the array, call an installed control plane, open a network connection, search the filesystem, retrieve a secret, or expose a captured command string. `READY FOR REVIEW` means that the captured evidence is internally consistent enough for human review. It does not mean approved, current, or executed.
manifest.json json View source
{
"name": "llmops-component-triage",
"release": "0.1.0-teaching",
"evidence_schema_versions": [1],
"minimum_python": "3.10",
"capabilities": ["read_fixture", "validate", "correlate", "render"],
"excluded_capabilities": ["network", "subprocess", "shell", "secret", "filesystem_discovery", "package_manager", "llmops_execution", "lifecycle_mutation"],
"files": {
"README.md": "4eefaf896151ad72427c5196bedc16cc670bf95de355930f7a1d621e14ae97e9",
"SKILL.md": "fabd7ea38f8aefac0d79490a9fa1936e37e67157fbc83c65396a1cc649a4a7c1",
"fixtures/complete.json": "f6e9f3ebaff00f16ef9e43fc729560252025614ad3842fd5cf7907bf02f9add7",
"fixtures/expected-complete.md": "34825231442c80f13909b8a316a25fc7e1ff15bdb31250955779670141b4c1b8",
"fixtures/stale.json": "79c13c4f93e8213d9a8b295d4a0ca483809877f5d0802d9252ebb82d5d350ded",
"fixtures/unobserved.json": "d7f8ad4c32409bdf27357fd1356ea531f65737c1d55e0a0b7fb6be5956686dbb",
"run_lab.py": "477de0932c65917ffb874b01c1c2413f0d2a109e850f4acf227414aaf6c33285",
"schemas/evidence-envelope.schema.json": "82cd9764f825369402fe0dcd86cfb66e936cede3299367c72b762b008c5000be",
"scripts/triage_packet.py": "23100884dc1b412a40932e86eb9b57f88ce25b38b194734d75086481c658c5cf",
"test_triage_packet.py": "0cafb5008a1dfe913770582c73a533f92ab8d217cfe9cbff067e6b2bd44bef62"
}
}
evidence-envelope.schema.json json View source
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://example.invalid/llmops-component-triage/evidence-envelope.schema.json",
"title": "Sanitized component triage evidence envelope",
"type": "object",
"additionalProperties": false,
"required": ["schema_version", "capture", "request", "status", "target_plan", "cascade_plan"],
"properties": {
"schema_version": {"const": 1},
"capture": {
"type": "object", "additionalProperties": false,
"required": ["captured_at", "toolkit_version", "config_hash", "catalog_hash"],
"properties": {
"captured_at": {"type": "string", "pattern": "^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z$"},
"toolkit_version": {"type": "string", "pattern": "^[a-z][a-z0-9-]{0,31}$"},
"config_hash": {"type": "string", "pattern": "^[a-f0-9]{64}$"},
"catalog_hash": {"type": "string", "pattern": "^[a-f0-9]{64}$"}
}
},
"request": {
"type": "object", "additionalProperties": false,
"required": ["component", "action", "cascade"],
"properties": {"component": {"$ref": "#/$defs/component"}, "action": {"const": "restart"}, "cascade": {"const": true}}
},
"status": {"type": "array", "items": {"$ref": "#/$defs/status"}},
"target_plan": {"type": "array", "items": {"$ref": "#/$defs/operation"}},
"cascade_plan": {"type": "array", "items": {"$ref": "#/$defs/operation"}}
},
"$defs": {
"component": {"type": "string", "pattern": "^[a-z][a-z0-9-]{0,31}:[a-z][a-z0-9-]{0,31}$"},
"symbol": {"type": "string", "pattern": "^[a-z][a-z0-9-]{0,31}$"},
"status": {
"type": "object", "additionalProperties": false,
"required": ["lifecycle", "desired_lifecycle", "health", "condition", "observability", "component", "host", "execution_user", "driver", "component_version", "toolkit_version", "config_hash", "catalog_hash", "error"],
"properties": {
"lifecycle": {"enum": ["running", "stopped", "unknown"]}, "desired_lifecycle": {"enum": ["running", "stopped"]},
"health": {"enum": ["healthy", "unhealthy", "unknown", "not-applicable"]}, "condition": {"$ref": "#/$defs/symbol"},
"observability": {"enum": ["observed", "unreachable", "unobserved"]}, "component": {"$ref": "#/$defs/component"},
"host": {"$ref": "#/$defs/symbol"}, "execution_user": {"$ref": "#/$defs/symbol"}, "driver": {"$ref": "#/$defs/symbol"},
"component_version": {"$ref": "#/$defs/symbol"}, "toolkit_version": {"$ref": "#/$defs/symbol"},
"config_hash": {"type": "string", "pattern": "^[a-f0-9]{64}$"}, "catalog_hash": {"type": "string", "pattern": "^[a-f0-9]{64}$"}, "error": {"type": "string", "maxLength": 80, "pattern": "^$|^[a-z][a-z0-9-]{0,31}$"}
}
},
"operation": {
"type": "object", "additionalProperties": false,
"required": ["component", "host", "driver", "action", "command"],
"properties": {"component": {"$ref": "#/$defs/component"}, "host": {"$ref": "#/$defs/symbol"}, "driver": {"$ref": "#/$defs/symbol"}, "action": {"enum": ["stop", "restart", "start"]}, "command": {"const": "<redacted-by-capture-adapter>"}}
}
}
}
triage_packet.py python View source
#!/usr/bin/env python3
"""Render a fixture-backed component restart review packet without execution."""
from __future__ import annotations
import argparse
import json
import re
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
SCHEMA_VERSION = 1
MAX_AGE_SECONDS = 600
REDACTED_COMMAND = "<redacted-by-capture-adapter>"
COMPONENT = re.compile(r"^[a-z][a-z0-9-]{0,31}:[a-z][a-z0-9-]{0,31}$")
SYMBOL = re.compile(r"^[a-z][a-z0-9-]{0,31}$")
HASH = re.compile(r"^[a-f0-9]{64}$")
UTC_TIME = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$")
UNSAFE = (
re.compile(r"(?:^|[\s=:,(])(?:/|~/)"),
re.compile(r"\b[a-z][a-z0-9+.-]*://", re.I),
re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b"),
re.compile(r"\b(?:password|token|secret|api[_-]?key)\s*[=:]", re.I),
re.compile(r"(?:bearer\s+[a-z0-9._-]+|BEGIN [A-Z ]*PRIVATE KEY)", re.I),
)
class PacketError(ValueError):
"""Bounded rejection that never reproduces unsafe input."""
def __init__(self, code: str, message: str):
super().__init__(f"{code}: {message}")
self.code = code
self.message = message
@dataclass(frozen=True)
class Packet:
decision: str
captured_at: str
target: str
target_status: dict[str, str]
cascade_stops: tuple[str, ...]
observed_running_impact: tuple[str, ...]
restore_order: tuple[str, ...]
gaps: tuple[str, ...]
operation_argv: tuple[str, ...]
def reject(code: str, message: str) -> None:
raise PacketError(code, message)
def exact_keys(value: Any, required: set[str], *, where: str) -> dict[str, Any]:
if not isinstance(value, dict) or set(value) != required:
reject("E_SCHEMA", f"{where} has missing or unknown fields")
return value
def reject_unsafe(value: Any) -> None:
if isinstance(value, dict):
for item in value.values():
reject_unsafe(item)
elif isinstance(value, list):
for item in value:
reject_unsafe(item)
elif isinstance(value, str) and value != REDACTED_COMMAND:
if any(pattern.search(value) for pattern in UNSAFE):
reject("E_PRIVACY", "unsafe string rejected")
def parse_time(value: str) -> datetime:
if not isinstance(value, str) or not UTC_TIME.fullmatch(value):
reject("E_TIME", "capture time must be RFC 3339 UTC")
try:
return datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
except ValueError:
reject("E_TIME", "capture time is not a valid calendar value")
def validate(envelope: Any) -> dict[str, Any]:
root = exact_keys(envelope, {"schema_version", "capture", "request", "status", "target_plan", "cascade_plan"}, where="envelope")
if type(root["schema_version"]) is not int or root["schema_version"] != SCHEMA_VERSION:
reject("E_VERSION", "unsupported evidence schema")
capture = exact_keys(root["capture"], {"captured_at", "toolkit_version", "config_hash", "catalog_hash"}, where="capture")
parse_time(capture["captured_at"])
if not isinstance(capture["toolkit_version"], str) or not SYMBOL.fullmatch(capture["toolkit_version"]):
reject("E_IDENTITY", "invalid toolkit identity")
if not isinstance(capture["config_hash"], str) or not isinstance(capture["catalog_hash"], str) or not HASH.fullmatch(capture["config_hash"]) or not HASH.fullmatch(capture["catalog_hash"]):
reject("E_IDENTITY", "invalid authority identity")
request = exact_keys(root["request"], {"component", "action", "cascade"}, where="request")
if not isinstance(request["component"], str) or not COMPONENT.fullmatch(request["component"]) or not isinstance(request["action"], str) or request["action"] != "restart" or request["cascade"] is not True:
reject("E_REQUEST", "unsupported request")
if not isinstance(root["status"], list) or not isinstance(root["target_plan"], list) or not isinstance(root["cascade_plan"], list):
reject("E_SCHEMA", "capture arrays are required")
reject_unsafe(root)
return root
def validate_status(rows: list[Any], capture: dict[str, str]) -> dict[str, dict[str, str]]:
required = {"lifecycle", "desired_lifecycle", "health", "condition", "observability", "component", "host", "execution_user", "driver", "component_version", "toolkit_version", "config_hash", "catalog_hash", "error"}
allowed_lifecycle = {"running", "stopped", "unknown"}
allowed_desired = {"running", "stopped"}
allowed_health = {"healthy", "unhealthy", "unknown", "not-applicable"}
allowed_observability = {"observed", "unreachable", "unobserved"}
found: dict[str, dict[str, str]] = {}
for raw in rows:
row = exact_keys(raw, required, where="status row")
string_fields = ("lifecycle", "desired_lifecycle", "health", "condition", "observability", "component", "host", "execution_user", "driver", "component_version", "toolkit_version", "config_hash", "catalog_hash", "error")
if any(not isinstance(row[field], str) for field in string_fields):
reject("E_STATUS", "status fields have invalid JSON types")
component = row["component"]
if not COMPONENT.fullmatch(component) or component in found:
reject("E_STATUS", "invalid or duplicate component observation")
if not SYMBOL.fullmatch(row["host"]) or not SYMBOL.fullmatch(row["execution_user"]) or not SYMBOL.fullmatch(row["driver"]):
reject("E_STATUS", "invalid symbolic status identity")
if row["lifecycle"] not in allowed_lifecycle or row["health"] not in allowed_health or row["observability"] not in allowed_observability:
reject("E_STATUS", "invalid status vocabulary")
if row["desired_lifecycle"] not in allowed_desired:
reject("E_STATUS", "invalid desired lifecycle")
if not SYMBOL.fullmatch(row["condition"]) or not SYMBOL.fullmatch(row["component_version"]):
reject("E_STATUS", "invalid symbolic status value")
if not isinstance(row["error"], str) or (row["error"] and not SYMBOL.fullmatch(row["error"])):
reject("E_STATUS", "invalid sanitized error code")
if len(row["error"]) > 80:
reject("E_STATUS", "sanitized error code is too long")
if any(row[key] != capture[key] for key in ("toolkit_version", "config_hash", "catalog_hash")):
reject("E_IDENTITY", "status and envelope identities differ")
found[component] = row
return found
def validate_plan(rows: list[Any]) -> list[dict[str, str]]:
required = {"component", "host", "driver", "action", "command"}
result: list[dict[str, str]] = []
seen: set[tuple[str, str]] = set()
for raw in rows:
row = exact_keys(raw, required, where="plan row")
if any(not isinstance(row[field], str) for field in required):
reject("E_PLAN", "plan fields have invalid JSON types")
if not COMPONENT.fullmatch(row["component"]) or not SYMBOL.fullmatch(row["host"]) or not SYMBOL.fullmatch(row["driver"]):
reject("E_PLAN", "invalid symbolic plan identity")
if row["action"] not in {"stop", "restart", "start"} or row["command"] != REDACTED_COMMAND:
reject("E_PLAN", "invalid plan operation")
identity = (row["component"], row["action"])
if identity in seen:
reject("E_PLAN", "duplicate plan operation")
seen.add(identity)
result.append(row)
return result
def build_packet(envelope: Any, *, as_of: str) -> Packet:
root = validate(envelope)
capture = root["capture"]
captured_at = parse_time(capture["captured_at"])
now = parse_time(as_of)
if now < captured_at:
reject("E_TIME", "as-of time precedes capture")
status = validate_status(root["status"], capture)
target_plan = validate_plan(root["target_plan"])
cascade = validate_plan(root["cascade_plan"])
target = root["request"]["component"]
if [(row["component"], row["action"]) for row in target_plan] != [(target, "restart")]:
reject("E_PLAN", "target plan must contain one target restart")
restart_positions = [index for index, row in enumerate(cascade) if row["action"] == "restart"]
if len(restart_positions) != 1 or cascade[restart_positions[0]]["component"] != target:
reject("E_PLAN", "cascade must contain one target restart")
pivot = restart_positions[0]
if any(row["action"] != "stop" for row in cascade[:pivot]) or any(row["action"] != "start" for row in cascade[pivot + 1:]):
reject("E_PLAN", "cascade phases are out of order")
stops = tuple(row["component"] for row in cascade[:pivot])
restores = tuple(row["component"] for row in cascade[pivot + 1:])
if restores != tuple(reversed(stops)) or target in stops:
reject("E_PLAN", "cascade membership is inconsistent")
target_identity = (target_plan[0]["component"], target_plan[0]["host"], target_plan[0]["driver"])
cascade_target_identity = (cascade[pivot]["component"], cascade[pivot]["host"], cascade[pivot]["driver"])
if target_identity != cascade_target_identity:
reject("E_PLAN", "target plan identities disagree")
for row in (*target_plan, *cascade):
observation = status.get(row["component"])
if observation is not None and (row["host"], row["driver"]) != (observation["host"], observation["driver"]):
reject("E_PLAN", "plan and status identities disagree")
gaps: list[str] = []
for component in (target, *stops):
row = status.get(component)
if row is None:
gaps.append(f"missing observation: {component}")
elif row["observability"] != "observed" or row["lifecycle"] == "unknown":
gaps.append(f"incomplete observation: {component}")
stale = (now - captured_at).total_seconds() > MAX_AGE_SECONDS
decision = "evidence_stale" if stale else "evidence_incomplete" if gaps else "ready_for_review"
observed = tuple(component for component in stops if status.get(component, {}).get("lifecycle") == "running" and status.get(component, {}).get("observability") == "observed")
operation = ("llmops", "component", "restart", target, "--cascade") if decision == "ready_for_review" else ()
target_row = status.get(target, {})
return Packet(decision, capture["captured_at"], target, {key: str(target_row.get(key, "unknown")) for key in ("lifecycle", "health", "observability")}, stops, observed, restores, tuple(gaps), operation)
def render_markdown(packet: Packet) -> str:
label = "READY FOR REVIEW" if packet.decision == "ready_for_review" else "NOT READY"
lines = ["# Component Restart Review", "", f"Decision: {label}", f"Evidence state: {packet.decision}", f"Evidence captured: {packet.captured_at.replace('T', ' ').replace('Z', ' UTC')}", f"Target: {packet.target}", "Requested scope: cascade restart", "", "## Observed target", "", f"- Lifecycle: {packet.target_status['lifecycle']}", f"- Health: {packet.target_status['health']}", f"- Observability: {packet.target_status['observability']}", "", "## Canonical impact", "", f"- Target-only plan: restart {packet.target}", f"- Cascade stop set: {', '.join(packet.cascade_stops) or 'none'}", f"- Observed running components within cascade impact: {', '.join(packet.observed_running_impact) or 'none'}", f"- Restore order: {', '.join(packet.restore_order) or 'none'}"]
if packet.gaps:
lines += ["", "## Evidence gaps", ""] + [f"- {gap}" for gap in packet.gaps]
if packet.operation_argv:
lines += ["", "## Equivalent CLI argument array", "", json.dumps(list(packet.operation_argv))]
lines += ["", "## Approval boundary", "", f"No command was executed. Captured status may have changed since {packet.captured_at.replace('T', ' ').replace('Z', ' UTC')}. Re-plan and re-observe through a separately qualified live adapter before any approved mutation.", ""]
return "\n".join(lines)
def render_json(packet: Packet) -> str:
return json.dumps({"decision": packet.decision, "captured_at": packet.captured_at, "target": packet.target, "target_status": packet.target_status, "cascade_stop_set": list(packet.cascade_stops), "observed_running_impact": list(packet.observed_running_impact), "restore_order": list(packet.restore_order), "evidence_gaps": list(packet.gaps), "operation_argv": list(packet.operation_argv), "executed": False}, indent=2, sort_keys=True) + "\n"
def load(path: Path) -> Any:
return json.loads(path.read_text(encoding="utf-8"))
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("fixture", type=Path)
parser.add_argument("--as-of", required=True)
parser.add_argument("--format", choices=("markdown", "json"), default="markdown")
args = parser.parse_args()
try:
packet = build_packet(load(args.fixture), as_of=args.as_of)
except (OSError, json.JSONDecodeError, PacketError) as exc:
if isinstance(exc, PacketError):
print(json.dumps({"decision": "input_rejected", "error_code": exc.code, "message": exc.message}, sort_keys=True))
else:
print(json.dumps({"decision": "input_rejected", "error_code": "E_INPUT", "message": "fixture could not be read"}, sort_keys=True))
return 2
print(render_markdown(packet) if args.format == "markdown" else render_json(packet), end="")
return 0
if __name__ == "__main__":
raise SystemExit(main())
run_lab.py python View source
from __future__ import annotations
import json
import hashlib
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parent
sys.path.insert(0, str(ROOT / "scripts"))
from triage_packet import build_packet, load, render_json, render_markdown # noqa: E402
AS_OF = "2030-01-02T03:09:05Z"
def main() -> int:
complete = build_packet(load(ROOT / "fixtures" / "complete.json"), as_of=AS_OF)
unobserved = build_packet(load(ROOT / "fixtures" / "unobserved.json"), as_of=AS_OF)
stale = build_packet(load(ROOT / "fixtures" / "stale.json"), as_of=AS_OF)
manifest = json.loads((ROOT / "manifest.json").read_text())
manifest_valid = all(hashlib.sha256((ROOT / relative).read_bytes()).hexdigest() == expected for relative, expected in manifest["files"].items())
checks = {
"complete evidence is ready for review": complete.decision == "ready_for_review",
"complete Markdown matches the golden file": render_markdown(complete) == (ROOT / "fixtures" / "expected-complete.md").read_text(),
"JSON preserves the review decision": json.loads(render_json(complete))["decision"] == "ready_for_review",
"target-only and cascade facts remain distinct": complete.cascade_stops == ("demo-stack:agent",),
"only observed running cascade members are reported": complete.observed_running_impact == ("demo-stack:agent",),
"the operation is a closed argument array": complete.operation_argv == ("llmops", "component", "restart", "demo-stack:model-proxy", "--cascade"),
"unobserved impact is not ready": unobserved.decision == "evidence_incomplete",
"incomplete evidence suppresses the operation array": not unobserved.operation_argv,
"stale evidence is not ready": stale.decision == "evidence_stale",
"stale evidence suppresses the operation array": not stale.operation_argv,
"manifest checksums match package files": manifest_valid,
"manifest capability boundary is closed": manifest["capabilities"] == ["read_fixture", "validate", "correlate", "render"],
"rendering creates no output directory": not (ROOT / "output").exists(),
"Python bytecode is absent": not any(ROOT.rglob("__pycache__")) and not any(ROOT.rglob("*.pyc")),
}
for label, passed in checks.items():
print(f"{'PASS' if passed else 'FAIL'}: {label}")
print(f"{sum(checks.values())}/{len(checks)} conditions passed")
return 0 if all(checks.values()) else 1
if __name__ == "__main__":
raise SystemExit(main())
test_triage_packet.py python View source
from __future__ import annotations
import ast
import contextlib
import copy
import hashlib
import io
import json
import re
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock
ROOT = Path(__file__).resolve().parent
sys.path.insert(0, str(ROOT / "scripts"))
from triage_packet import PacketError, build_packet, load, main, render_json, render_markdown # noqa: E402
AS_OF = "2030-01-02T03:09:05Z"
def audit_read_only_source(source: str) -> None:
tree = ast.parse(source)
forbidden_roots = {"subprocess", "socket", "requests", "urllib", "http", "paramiko", "llmops_kit"}
forbidden_names = {"eval", "exec", "compile", "__import__"}
filesystem_mutations = {"write_text", "write_bytes", "mkdir", "unlink", "rename", "replace", "touch", "rmdir", "symlink_to", "hardlink_to"}
aliases: dict[str, str] = {}
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for item in node.names:
aliases[item.asname or item.name.split(".")[0]] = item.name
if item.name.split(".")[0] in forbidden_roots:
raise AssertionError("forbidden import")
elif isinstance(node, ast.ImportFrom):
module = node.module or ""
if module.split(".")[0] in forbidden_roots or module == "os" and any(item.name in {"system", "popen"} for item in node.names):
raise AssertionError("forbidden direct import")
for item in node.names:
aliases[item.asname or item.name] = f"{module}.{item.name}".strip(".")
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
if isinstance(node.func, ast.Name):
name = node.func.id
path = aliases.get(name, name)
if name in forbidden_names or path.split(".")[0] in forbidden_roots or path in {"os.system", "os.popen"}:
raise AssertionError("forbidden call")
elif isinstance(node.func, ast.Attribute):
path_receiver = isinstance(node.func.value, ast.Name) and aliases.get(node.func.value.id, node.func.value.id) == "pathlib.Path"
path_receiver = path_receiver or isinstance(node.func.value, ast.Call) and isinstance(node.func.value.func, ast.Name) and aliases.get(node.func.value.func.id, node.func.value.func.id) == "pathlib.Path"
if node.func.attr in filesystem_mutations and (node.func.attr != "replace" or path_receiver):
raise AssertionError("filesystem mutation")
parts = [node.func.attr]
value = node.func.value
while isinstance(value, ast.Attribute):
parts.append(value.attr)
value = value.value
if isinstance(value, ast.Name):
parts.append(aliases.get(value.id, value.id))
path = ".".join(reversed(parts))
if path.split(".")[0] in forbidden_roots or path in {"os.system", "os.popen"}:
raise AssertionError("forbidden call")
class TriagePacketTests(unittest.TestCase):
def fixture(self, name: str = "complete.json"):
return load(ROOT / "fixtures" / name)
def rejected(self, envelope, code: str):
with self.assertRaises(PacketError) as caught:
build_packet(envelope, as_of=AS_OF)
self.assertEqual(caught.exception.code, code)
def invoke_cli(self, envelope):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "fixture.json"
path.write_text(json.dumps(envelope))
output = io.StringIO()
with mock.patch.object(sys, "argv", ["triage_packet.py", str(path), "--as-of", AS_OF]), contextlib.redirect_stdout(output):
status = main()
return status, output.getvalue()
def test_complete_is_ready(self):
self.assertEqual(build_packet(self.fixture(), as_of=AS_OF).decision, "ready_for_review")
def test_golden_markdown(self):
packet = build_packet(self.fixture(), as_of=AS_OF)
self.assertEqual(render_markdown(packet), (ROOT / "fixtures" / "expected-complete.md").read_text())
def test_rendering_is_deterministic(self):
packet = build_packet(self.fixture(), as_of=AS_OF)
self.assertEqual(render_json(packet), render_json(packet))
def test_json_and_markdown_agree(self):
packet = build_packet(self.fixture(), as_of=AS_OF)
structured = json.loads(render_json(packet))
markdown = render_markdown(packet)
self.assertEqual(structured["decision"], packet.decision)
self.assertEqual(structured["captured_at"], packet.captured_at)
self.assertEqual(structured["target"], packet.target)
self.assertEqual(structured["target_status"], packet.target_status)
self.assertEqual(structured["cascade_stop_set"], list(packet.cascade_stops))
self.assertEqual(structured["observed_running_impact"], list(packet.observed_running_impact))
self.assertEqual(structured["restore_order"], list(packet.restore_order))
self.assertEqual(structured["evidence_gaps"], list(packet.gaps))
self.assertEqual(structured["operation_argv"], list(packet.operation_argv))
self.assertFalse(structured["executed"])
for value in (packet.captured_at.replace("T", " ").replace("Z", " UTC"), packet.target, *packet.target_status.values(), *packet.cascade_stops, *packet.observed_running_impact, *packet.restore_order, json.dumps(list(packet.operation_argv))):
self.assertIn(value, markdown)
def test_target_plan_is_target_only(self):
envelope = self.fixture()
envelope["target_plan"].append(copy.deepcopy(envelope["cascade_plan"][0]))
self.rejected(envelope, "E_PLAN")
def test_invalid_cascade_order_is_rejected(self):
envelope = self.fixture()
envelope["cascade_plan"][0]["action"] = "start"
self.rejected(envelope, "E_PLAN")
def test_two_dependent_restore_order_is_exact_reverse(self):
envelope = self.fixture()
second_status = copy.deepcopy(envelope["status"][-1])
second_status["component"] = "demo-stack:dashboard"
envelope["status"].append(second_status)
stop = copy.deepcopy(envelope["cascade_plan"][0])
stop["component"] = "demo-stack:dashboard"
start = copy.deepcopy(stop)
start["action"] = "start"
envelope["cascade_plan"] = [stop, envelope["cascade_plan"][0], envelope["cascade_plan"][1], envelope["cascade_plan"][2], start]
packet = build_packet(envelope, as_of=AS_OF)
self.assertEqual(packet.restore_order, tuple(reversed(packet.cascade_stops)))
envelope["cascade_plan"][-2:] = list(reversed(envelope["cascade_plan"][-2:]))
self.rejected(envelope, "E_PLAN")
def test_impact_comes_from_cascade_membership(self):
packet = build_packet(self.fixture(), as_of=AS_OF)
self.assertEqual(packet.observed_running_impact, ("demo-stack:agent",))
self.assertNotIn("demo-stack:model", packet.observed_running_impact)
def test_unobserved_is_incomplete_and_suppresses_operation(self):
packet = build_packet(self.fixture("unobserved.json"), as_of=AS_OF)
self.assertEqual(packet.decision, "evidence_incomplete")
self.assertEqual(packet.operation_argv, ())
def test_missing_affected_status_is_incomplete(self):
envelope = self.fixture()
envelope["status"] = [row for row in envelope["status"] if row["component"] != "demo-stack:agent"]
self.assertEqual(build_packet(envelope, as_of=AS_OF).decision, "evidence_incomplete")
def test_stale_is_not_ready(self):
packet = build_packet(self.fixture("stale.json"), as_of=AS_OF)
self.assertEqual(packet.decision, "evidence_stale")
self.assertEqual(packet.operation_argv, ())
def test_mismatched_identity_is_rejected(self):
envelope = self.fixture()
envelope["status"][0]["config_hash"] = "c" * 64
self.rejected(envelope, "E_IDENTITY")
def test_plan_host_driver_and_target_identities_are_correlated(self):
for section, index, field in (("cascade_plan", 0, "host"), ("cascade_plan", 0, "driver"), ("target_plan", 0, "host")):
envelope = self.fixture()
envelope[section][index][field] = "other-node" if field == "host" else "launchd"
self.rejected(envelope, "E_PLAN")
envelope = self.fixture()
envelope["cascade_plan"][1]["host"] = "other-node"
self.rejected(envelope, "E_PLAN")
def test_unknown_field_is_rejected(self):
envelope = self.fixture()
envelope["request"]["extra"] = "value"
self.rejected(envelope, "E_SCHEMA")
def test_missing_field_is_rejected(self):
envelope = self.fixture()
del envelope["capture"]["catalog_hash"]
self.rejected(envelope, "E_SCHEMA")
def test_duplicate_observation_is_rejected(self):
envelope = self.fixture()
envelope["status"].append(copy.deepcopy(envelope["status"][0]))
self.rejected(envelope, "E_STATUS")
def test_duplicate_operation_is_rejected(self):
envelope = self.fixture()
envelope["cascade_plan"].insert(1, copy.deepcopy(envelope["cascade_plan"][0]))
self.rejected(envelope, "E_PLAN")
def test_unsafe_material_is_rejected_without_echo(self):
for unsafe in ("/private/example", "failure path=/private/example", "https://example.invalid", "failure ftp://private.invalid", "192.0.2.10", "token=example", "secret: example", "Bearer example"):
envelope = self.fixture()
envelope["status"][0]["error"] = unsafe
with self.assertRaises(PacketError) as caught:
build_packet(envelope, as_of=AS_OF)
self.assertEqual(caught.exception.code, "E_PRIVACY")
self.assertNotIn(unsafe, str(caught.exception))
def test_runtime_enforces_every_published_status_constraint(self):
for field, value in (("desired_lifecycle", "unknown"), ("condition", "not_valid"), ("component_version", "not_valid"), ("error", "x" * 81)):
envelope = self.fixture()
envelope["status"][0][field] = value
self.rejected(envelope, "E_STATUS")
def test_malformed_json_types_and_calendar_values_are_bounded_at_cli(self):
cases = []
integer_hash = self.fixture()
integer_hash["capture"]["config_hash"] = int("1" * 64)
for row in integer_hash["status"]:
row["config_hash"] = int("1" * 64)
cases.append((integer_hash, "E_IDENTITY"))
for field in ("lifecycle", "desired_lifecycle", "health", "observability"):
envelope = self.fixture()
envelope["status"][0][field] = ["running"]
cases.append((envelope, "E_STATUS"))
plan_action = self.fixture()
plan_action["target_plan"][0]["action"] = ["restart"]
cases.append((plan_action, "E_PLAN"))
invalid_date = self.fixture()
invalid_date["capture"]["captured_at"] = "2030-02-30T03:04:05Z"
cases.append((invalid_date, "E_TIME"))
for envelope, code in cases:
status, output = self.invoke_cli(envelope)
self.assertEqual(status, 2)
self.assertEqual(json.loads(output)["error_code"], code)
self.assertNotIn("Traceback", output)
def test_raw_command_is_rejected(self):
envelope = self.fixture()
envelope["target_plan"][0]["command"] = "llmops component restart example"
self.rejected(envelope, "E_PLAN")
def test_unsupported_version_action_and_boolean_version_are_rejected(self):
for value, code in ((2, "E_VERSION"), (True, "E_VERSION")):
envelope = self.fixture()
envelope["schema_version"] = value
self.rejected(envelope, code)
envelope = self.fixture()
envelope["request"]["action"] = "stop"
self.rejected(envelope, "E_REQUEST")
def test_argument_array_is_closed(self):
packet = build_packet(self.fixture(), as_of=AS_OF)
self.assertEqual(packet.operation_argv, ("llmops", "component", "restart", "demo-stack:model-proxy", "--cascade"))
def test_source_has_no_execution_surface(self):
source = (ROOT / "scripts" / "triage_packet.py").read_text()
audit_read_only_source(source)
def test_ast_audit_rejects_direct_alias_and_filesystem_bypasses(self):
bypasses = (
"from subprocess import run\nrun([])\n",
"from os import system as invoke\ninvoke('example')\n",
"from pathlib import Path\nPath('example').write_text('value')\n",
)
for source in bypasses:
with self.assertRaises(AssertionError):
audit_read_only_source(source)
def test_fixture_and_golden_content_has_no_private_identifier(self):
patterns = [re.compile(r"/Users/|/home/|~/"), re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b"), re.compile(r"BEGIN [A-Z ]*PRIVATE KEY"), re.compile(r"\b(?:password|token|secret|api[_-]?key)\s*[=:]", re.I)]
for path in [*sorted((ROOT / "fixtures").glob("*"))]:
text = path.read_text()
self.assertFalse(any(pattern.search(text) for pattern in patterns), path.name)
def test_manifest_capabilities_and_checksums(self):
manifest = json.loads((ROOT / "manifest.json").read_text())
self.assertEqual(manifest["capabilities"], ["read_fixture", "validate", "correlate", "render"])
self.assertIn("lifecycle_mutation", manifest["excluded_capabilities"])
for relative, expected in manifest["files"].items():
self.assertEqual(hashlib.sha256((ROOT / relative).read_bytes()).hexdigest(), expected)
if __name__ == "__main__":
unittest.main()
complete.json json View source
{
"schema_version": 1,
"capture": {"captured_at": "2030-01-02T03:04:05Z", "toolkit_version": "example", "config_hash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "catalog_hash": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"},
"request": {"component": "demo-stack:model-proxy", "action": "restart", "cascade": true},
"status": [
{"lifecycle": "running", "desired_lifecycle": "running", "health": "healthy", "condition": "ready", "observability": "observed", "component": "demo-stack:model", "host": "model-node", "execution_user": "service-user", "driver": "process", "component_version": "example", "toolkit_version": "example", "config_hash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "catalog_hash": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "error": ""},
{"lifecycle": "running", "desired_lifecycle": "running", "health": "healthy", "condition": "ready", "observability": "observed", "component": "demo-stack:model-proxy", "host": "model-node", "execution_user": "service-user", "driver": "process", "component_version": "example", "toolkit_version": "example", "config_hash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "catalog_hash": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "error": ""},
{"lifecycle": "running", "desired_lifecycle": "running", "health": "healthy", "condition": "ready", "observability": "observed", "component": "demo-stack:agent", "host": "agent-node", "execution_user": "service-user", "driver": "process", "component_version": "example", "toolkit_version": "example", "config_hash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "catalog_hash": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "error": ""}
],
"target_plan": [
{"component": "demo-stack:model-proxy", "host": "model-node", "driver": "process", "action": "restart", "command": "<redacted-by-capture-adapter>"}
],
"cascade_plan": [
{"component": "demo-stack:agent", "host": "agent-node", "driver": "process", "action": "stop", "command": "<redacted-by-capture-adapter>"},
{"component": "demo-stack:model-proxy", "host": "model-node", "driver": "process", "action": "restart", "command": "<redacted-by-capture-adapter>"},
{"component": "demo-stack:agent", "host": "agent-node", "driver": "process", "action": "start", "command": "<redacted-by-capture-adapter>"}
]
}
unobserved.json json View source
{
"schema_version": 1,
"capture": {"captured_at": "2030-01-02T03:04:05Z", "toolkit_version": "example", "config_hash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "catalog_hash": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"},
"request": {"component": "demo-stack:model-proxy", "action": "restart", "cascade": true},
"status": [
{"lifecycle": "running", "desired_lifecycle": "running", "health": "healthy", "condition": "ready", "observability": "observed", "component": "demo-stack:model-proxy", "host": "model-node", "execution_user": "service-user", "driver": "process", "component_version": "example", "toolkit_version": "example", "config_hash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "catalog_hash": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "error": ""},
{"lifecycle": "unknown", "desired_lifecycle": "running", "health": "unknown", "condition": "unknown", "observability": "unreachable", "component": "demo-stack:agent", "host": "agent-node", "execution_user": "service-user", "driver": "process", "component_version": "example", "toolkit_version": "example", "config_hash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "catalog_hash": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "error": "bounded-observation-unavailable"}
],
"target_plan": [{"component": "demo-stack:model-proxy", "host": "model-node", "driver": "process", "action": "restart", "command": "<redacted-by-capture-adapter>"}],
"cascade_plan": [
{"component": "demo-stack:agent", "host": "agent-node", "driver": "process", "action": "stop", "command": "<redacted-by-capture-adapter>"},
{"component": "demo-stack:model-proxy", "host": "model-node", "driver": "process", "action": "restart", "command": "<redacted-by-capture-adapter>"},
{"component": "demo-stack:agent", "host": "agent-node", "driver": "process", "action": "start", "command": "<redacted-by-capture-adapter>"}
]
}
stale.json json View source
{
"schema_version": 1,
"capture": {"captured_at": "2030-01-02T02:00:00Z", "toolkit_version": "example", "config_hash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "catalog_hash": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"},
"request": {"component": "demo-stack:model-proxy", "action": "restart", "cascade": true},
"status": [
{"lifecycle": "running", "desired_lifecycle": "running", "health": "healthy", "condition": "ready", "observability": "observed", "component": "demo-stack:model-proxy", "host": "model-node", "execution_user": "service-user", "driver": "process", "component_version": "example", "toolkit_version": "example", "config_hash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "catalog_hash": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "error": ""},
{"lifecycle": "running", "desired_lifecycle": "running", "health": "healthy", "condition": "ready", "observability": "observed", "component": "demo-stack:agent", "host": "agent-node", "execution_user": "service-user", "driver": "process", "component_version": "example", "toolkit_version": "example", "config_hash": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "catalog_hash": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "error": ""}
],
"target_plan": [{"component": "demo-stack:model-proxy", "host": "model-node", "driver": "process", "action": "restart", "command": "<redacted-by-capture-adapter>"}],
"cascade_plan": [
{"component": "demo-stack:agent", "host": "agent-node", "driver": "process", "action": "stop", "command": "<redacted-by-capture-adapter>"},
{"component": "demo-stack:model-proxy", "host": "model-node", "driver": "process", "action": "restart", "command": "<redacted-by-capture-adapter>"},
{"component": "demo-stack:agent", "host": "agent-node", "driver": "process", "action": "start", "command": "<redacted-by-capture-adapter>"}
]
}
expected-complete.md markdown View source
# Component Restart Review
Decision: READY FOR REVIEW
Evidence state: ready_for_review
Evidence captured: 2030-01-02 03:04:05 UTC
Target: demo-stack:model-proxy
Requested scope: cascade restart
## Observed target
- Lifecycle: running
- Health: healthy
- Observability: observed
## Canonical impact
- Target-only plan: restart demo-stack:model-proxy
- Cascade stop set: demo-stack:agent
- Observed running components within cascade impact: demo-stack:agent
- Restore order: demo-stack:agent
## Equivalent CLI argument array
["llmops", "component", "restart", "demo-stack:model-proxy", "--cascade"]
## Approval boundary
No command was executed. Captured status may have changed since 2030-01-02 03:04:05 UTC. Re-plan and re-observe through a separately qualified live adapter before any approved mutation.
2. Verify and Extract the Archive
Download the reviewed archive and its checksum, then verify it before extraction:
shasum -a 256 -c llmops-component-triage.zip.sha256
mkdir triage-lab
cd triage-lab
unzip ../llmops-component-triage.zip
cd llmops-component-triage
python3 --version
The package requires Python 3.10 or newer and uses only the standard library. There is nothing to install with pip.
3. Read the Skill Contract
Open SKILL.md before the Python. It says when the skill applies, what evidence it accepts, what it refuses to infer, and where it stops. The schema and renderer live outside the prompt-facing instructions because validation should not depend on a model remembering every field or regular expression.
The central rule is simple: the skill may correlate evidence, but it may not become a second dependency planner. The captured cascade plan supplies the affected membership. The status capture supplies observations. Their intersection tells us which members inside that canonical impact were observed running at capture time.
4. Render the Complete Packet
Use the fixed as-of time so freshness does not depend on the day you run the tutorial:
python3 -B scripts/triage_packet.py fixtures/complete.json --as-of 2030-01-02T03:09:05Z --format markdown
The complete fixture contains a target-only plan with one proxy restart and a cascade plan that stops an invented agent, restarts the proxy, and restores the agent. The status snapshot says that both affected components were observed and running. The output can therefore say READY FOR REVIEW and display this argument array:
["llmops", "component", "restart", "demo-stack:model-proxy", "--cascade"]
That is structured explanation, not shell text and not permission. The package never executes the array. It also warns that the observation may have changed and requires a separately qualified live adapter to re-plan and re-observe before any approved mutation.
5. Compare Markdown and JSON
The same packet is available as JSON:
python3 -B scripts/triage_packet.py fixtures/complete.json --as-of 2030-01-02T03:09:05Z --format json
Both formats carry the decision, target, observed target state, cascade stop set, observed running impact, restore order, evidence gaps, argument array, and executed: false boundary. The tests compare these facts rather than trusting two renderers that merely look similar.
6. Remove One Reliable Observation
Now run the unobserved fixture:
python3 -B scripts/triage_packet.py fixtures/unobserved.json --as-of 2030-01-02T03:09:05Z --format markdown
The captured canonical plan still includes the agent, but its observation is unreachable. That is valid but incomplete evidence. The packet reports NOT READY, names the gap, and suppresses the argument array. It does not turn “the planner selected it” into “I know its current state.”
7. Let Good Evidence Go Stale
Well-formed evidence is not timeless:
python3 -B scripts/triage_packet.py fixtures/stale.json --as-of 2030-01-02T03:09:05Z --format markdown
This capture is older than the lab’s fixed ten-minute window. It becomes evidence_stale, even though its schema, identities, plans, and observations remain internally consistent. Again, no argument array is offered.
8. Try the Rejection Tests
The unit suite mutates clean fixtures to introduce unknown fields, missing fields, mismatched hashes, duplicate observations, duplicate operations, invalid plan order, raw command strings, paths, addresses, URI authorities, credential assignments, bearer material, unsupported actions, and a boolean where the integer schema version belongs:
python3 -B -m unittest -v test_triage_packet.py
Rejected values are not echoed into the error packet. That detail matters because a validator that helpfully repeats a secret has already lost the privacy argument. These recognizable-form checks are defense in depth, not a promise to detect every arbitrary secret. A future capture adapter still owns sanitization before the envelope becomes model-visible.
9. Audit the Execution Surface
One test parses the renderer’s Python abstract syntax tree and rejects imports associated with subprocesses, sockets, HTTP, SSH, or LLM-Ops-Kit. It also checks for shell execution. The manifest separately declares the package’s four allowed capabilities.
This proves a bounded property of the published teaching package. It does not prove that Python, the reader’s computer, or some future collector cannot mutate anything. The claim is about these reviewed files and this fixture path.
10. Run the Acceptance Script
python3 -B run_lab.py
The acceptance runner checks fourteen end-to-end conditions. The separate unit suite currently contains twenty-seven tests. Both execute entirely in memory apart from reading the package files, and -B prevents Python bytecode from being written.
11. Clean Up
Return to the directory above the disposable lab and remove it:
cd ../..
rm -rf triage-lab
At the end of this walkthrough, no service has been queried, no installed control plane has run, no configuration has changed, no secret has been accessed, and no lifecycle operation has occurred.
Current State
The reviewed package is a model-free teaching artifact over three invented captures. Its complete case produces a deterministic approval packet, its incomplete and stale cases stop short of an operation array, and its tests exercise the closed schema, privacy boundary, identity binding, plan invariants, correlation rules, and lack of an execution surface. READY FOR REVIEW is still only a review state.
Next Work
A live collector would be a different artifact. It would need a pinned runtime identity, an explicit update policy, current schema negotiation, bounded transport, sanitization before model visibility, capture-time identity binding, race-aware revalidation, cancellation, timeouts, and its own side-effect tests. An executor would require another explicit approval boundary and would call the canonical operation API rather than treating this packet as shell text.
Join the Discussion
Comments for this post live in GitHub Discussions. That keeps moderation in one place and gives the conversation a stable home.