561 lines
22 KiB
Python
561 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
import shutil
|
|
import sys
|
|
from tempfile import TemporaryDirectory
|
|
from typing import Any
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
DEFAULT_AGENT_DIR = ROOT / "state" / "agents"
|
|
SCHEMA_VERSION = "ghostlight.agent_state.v0"
|
|
|
|
|
|
ROLE_TARGETS = {
|
|
"imagination": ("epiphany.imagination", "imagination.agent-state.json"),
|
|
"modeling": ("epiphany.body", "body.agent-state.json"),
|
|
"verification": ("epiphany.soul", "soul.agent-state.json"),
|
|
"implementation": ("epiphany.hands", "hands.agent-state.json"),
|
|
"reorientation": ("epiphany.life", "life.agent-state.json"),
|
|
"research": ("epiphany.eyes", "eyes.agent-state.json"),
|
|
"coordinator": ("epiphany.self", "self.agent-state.json"),
|
|
}
|
|
|
|
ALLOWED_PATCH_FIELDS = {
|
|
"agentId",
|
|
"reason",
|
|
"evidenceIds",
|
|
"semanticMemories",
|
|
"episodicMemories",
|
|
"relationshipMemories",
|
|
"goals",
|
|
"values",
|
|
"privateNotes",
|
|
}
|
|
|
|
FORBIDDEN_PATCH_FIELDS = {
|
|
"statePatch",
|
|
"objective",
|
|
"activeSubgoalId",
|
|
"subgoals",
|
|
"invariants",
|
|
"graphs",
|
|
"graphFrontier",
|
|
"graphCheckpoint",
|
|
"scratch",
|
|
"investigationCheckpoint",
|
|
"jobBindings",
|
|
"planning",
|
|
"churn",
|
|
"mode",
|
|
"codeEdits",
|
|
"files",
|
|
"authorityScope",
|
|
"backendJobId",
|
|
"rawResult",
|
|
}
|
|
|
|
|
|
def load_json(path: Path) -> Any:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def write_json(path: Path, value: Any) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(value, indent=2, sort_keys=False) + "\n", encoding="utf-8")
|
|
|
|
|
|
def role_target(role_id: str) -> tuple[str, Path]:
|
|
try:
|
|
agent_id, filename = ROLE_TARGETS[role_id]
|
|
except KeyError as exc:
|
|
raise ValueError(f"unknown role id: {role_id}") from exc
|
|
return agent_id, DEFAULT_AGENT_DIR / filename
|
|
|
|
|
|
def require_object(value: Any, path: str, errors: list[str]) -> dict[str, Any]:
|
|
if isinstance(value, dict):
|
|
return value
|
|
errors.append(f"{path} must be an object")
|
|
return {}
|
|
|
|
|
|
def require_array(value: Any, path: str, errors: list[str]) -> list[Any]:
|
|
if isinstance(value, list):
|
|
return value
|
|
errors.append(f"{path} must be an array")
|
|
return []
|
|
|
|
|
|
def check_unit(value: Any, path: str, errors: list[str]) -> None:
|
|
if not isinstance(value, (int, float)) or isinstance(value, bool) or not 0 <= float(value) <= 1:
|
|
errors.append(f"{path} must be a number between 0 and 1")
|
|
|
|
|
|
def check_string(value: Any, path: str, errors: list[str], *, max_len: int = 800) -> None:
|
|
if not isinstance(value, str) or not value.strip() or len(value) > max_len:
|
|
errors.append(f"{path} must be non-empty text under {max_len} characters")
|
|
|
|
|
|
def validate_memory(record: Any, path: str, errors: list[str]) -> None:
|
|
item = require_object(record, path, errors)
|
|
if not item:
|
|
return
|
|
check_string(item.get("memory_id"), f"{path}.memory_id", errors, max_len=120)
|
|
check_string(item.get("summary"), f"{path}.summary", errors, max_len=800)
|
|
check_unit(item.get("salience"), f"{path}.salience", errors)
|
|
check_unit(item.get("confidence"), f"{path}.confidence", errors)
|
|
|
|
|
|
def validate_agent_file(path: Path) -> list[str]:
|
|
errors: list[str] = []
|
|
try:
|
|
document = load_json(path)
|
|
except Exception as exc:
|
|
return [f"{path}: failed to parse JSON: {exc}"]
|
|
|
|
root = require_object(document, str(path), errors)
|
|
if root.get("schema_version") != SCHEMA_VERSION:
|
|
errors.append(f"{path}: schema_version must be {SCHEMA_VERSION!r}")
|
|
world = require_object(root.get("world"), f"{path}.world", errors)
|
|
for key in ("world_id", "setting", "canon_context"):
|
|
if key == "canon_context":
|
|
require_array(world.get(key), f"{path}.world.{key}", errors)
|
|
else:
|
|
check_string(world.get(key), f"{path}.world.{key}", errors)
|
|
time = require_object(world.get("time"), f"{path}.world.time", errors)
|
|
check_string(time.get("label"), f"{path}.world.time.label", errors, max_len=200)
|
|
|
|
agents = require_array(root.get("agents"), f"{path}.agents", errors)
|
|
if len(agents) != 1:
|
|
errors.append(f"{path}.agents must contain exactly one agent")
|
|
for index, raw_agent in enumerate(agents):
|
|
agent = require_object(raw_agent, f"{path}.agents[{index}]", errors)
|
|
check_string(agent.get("agent_id"), f"{path}.agents[{index}].agent_id", errors, max_len=120)
|
|
identity = require_object(agent.get("identity"), f"{path}.agents[{index}].identity", errors)
|
|
for key in ("name", "origin", "public_description"):
|
|
check_string(identity.get(key), f"{path}.agents[{index}].identity.{key}", errors)
|
|
require_array(identity.get("roles"), f"{path}.agents[{index}].identity.roles", errors)
|
|
|
|
canonical = require_object(
|
|
agent.get("canonical_state"), f"{path}.agents[{index}].canonical_state", errors
|
|
)
|
|
for key in (
|
|
"underlying_organization",
|
|
"stable_dispositions",
|
|
"behavioral_dimensions",
|
|
"presentation_strategy",
|
|
"voice_style",
|
|
"situational_state",
|
|
):
|
|
require_object(canonical.get(key), f"{path}.agents[{index}].canonical_state.{key}", errors)
|
|
values = require_array(
|
|
canonical.get("values"), f"{path}.agents[{index}].canonical_state.values", errors
|
|
)
|
|
for value_index, raw_value in enumerate(values):
|
|
value = require_object(
|
|
raw_value,
|
|
f"{path}.agents[{index}].canonical_state.values[{value_index}]",
|
|
errors,
|
|
)
|
|
check_string(value.get("value_id"), f"{path}.values[{value_index}].value_id", errors)
|
|
check_string(value.get("label"), f"{path}.values[{value_index}].label", errors)
|
|
check_unit(value.get("priority"), f"{path}.values[{value_index}].priority", errors)
|
|
if not isinstance(value.get("unforgivable_if_betrayed"), bool):
|
|
errors.append(f"{path}.values[{value_index}].unforgivable_if_betrayed must be boolean")
|
|
|
|
goals = require_array(agent.get("goals"), f"{path}.agents[{index}].goals", errors)
|
|
for goal_index, raw_goal in enumerate(goals):
|
|
goal = require_object(raw_goal, f"{path}.agents[{index}].goals[{goal_index}]", errors)
|
|
check_string(goal.get("goal_id"), f"{path}.goals[{goal_index}].goal_id", errors)
|
|
check_string(goal.get("description"), f"{path}.goals[{goal_index}].description", errors)
|
|
if goal.get("scope") not in {"immediate", "scene", "case", "arc", "life"}:
|
|
errors.append(f"{path}.goals[{goal_index}].scope is not a Ghostlight scope")
|
|
check_unit(goal.get("priority"), f"{path}.goals[{goal_index}].priority", errors)
|
|
check_string(goal.get("emotional_stake"), f"{path}.goals[{goal_index}].emotional_stake", errors)
|
|
if goal.get("status") not in {"active", "blocked", "dormant", "resolved", "abandoned"}:
|
|
errors.append(f"{path}.goals[{goal_index}].status is not a Ghostlight status")
|
|
|
|
memories = require_object(agent.get("memories"), f"{path}.agents[{index}].memories", errors)
|
|
for bundle_name in ("episodic", "semantic", "relationship_summaries"):
|
|
for memory_index, raw_memory in enumerate(
|
|
require_array(memories.get(bundle_name), f"{path}.memories.{bundle_name}", errors)
|
|
):
|
|
validate_memory(raw_memory, f"{path}.memories.{bundle_name}[{memory_index}]", errors)
|
|
|
|
require_array(
|
|
agent.get("perceived_state_overlays"),
|
|
f"{path}.agents[{index}].perceived_state_overlays",
|
|
errors,
|
|
)
|
|
|
|
for key in ("relationships", "events", "scenes"):
|
|
require_array(root.get(key), f"{path}.{key}", errors)
|
|
return errors
|
|
|
|
|
|
def valid_identifier(value: Any, prefix: str) -> bool:
|
|
return (
|
|
isinstance(value, str)
|
|
and value.startswith(prefix)
|
|
and len(value) <= 120
|
|
and all(ch.isascii() and (ch.isalnum() or ch in "-_.") for ch in value)
|
|
)
|
|
|
|
|
|
def review_patch(role_id: str, patch: Any) -> dict[str, Any]:
|
|
agent_id, target_path = role_target(role_id)
|
|
reasons: list[str] = []
|
|
if not isinstance(patch, dict):
|
|
reasons.append("selfPatch must be a JSON object")
|
|
else:
|
|
actual_agent = patch.get("agentId")
|
|
if actual_agent != agent_id:
|
|
reasons.append(f"selfPatch agentId {actual_agent!r} does not match this lane; expected {agent_id!r}")
|
|
reason = patch.get("reason")
|
|
if not isinstance(reason, str) or len(reason.strip()) < 16 or len(reason) > 800:
|
|
reasons.append("selfPatch reason must be a bounded explanation of at least 16 characters")
|
|
for key in patch:
|
|
if key in FORBIDDEN_PATCH_FIELDS:
|
|
reasons.append(
|
|
f"selfPatch field {key!r} is project truth or authority; use the proper Epiphany control surface instead"
|
|
)
|
|
elif key not in ALLOWED_PATCH_FIELDS:
|
|
reasons.append(f"selfPatch field {key!r} is not part of the bounded memory mutation contract")
|
|
|
|
mutation_count = 0
|
|
for field in ("semanticMemories", "episodicMemories", "relationshipMemories"):
|
|
mutation_count += review_memory_patch_array(patch, field, reasons)
|
|
mutation_count += review_goal_patch_array(patch, reasons)
|
|
mutation_count += review_value_patch_array(patch, reasons)
|
|
mutation_count += review_private_notes(patch, reasons)
|
|
review_string_array(patch, "evidenceIds", reasons, max_items=16, max_len=160)
|
|
if mutation_count == 0:
|
|
reasons.append(
|
|
"selfPatch must contain at least one semantic memory, episodic memory, relationship memory, goal, value, or private note"
|
|
)
|
|
|
|
return {
|
|
"status": "rejected" if reasons else "accepted",
|
|
"targetAgentId": agent_id,
|
|
"targetPath": str(target_path),
|
|
"reasons": reasons,
|
|
}
|
|
|
|
|
|
def review_memory_patch_array(patch: dict[str, Any], field: str, reasons: list[str]) -> int:
|
|
if field not in patch:
|
|
return 0
|
|
value = patch[field]
|
|
if not isinstance(value, list):
|
|
reasons.append(f"selfPatch {field} must be an array")
|
|
return 0
|
|
if len(value) > 8:
|
|
reasons.append(f"selfPatch {field} may contain at most 8 records")
|
|
for index, item in enumerate(value):
|
|
if not isinstance(item, dict):
|
|
reasons.append(f"selfPatch {field}[{index}] must be an object")
|
|
continue
|
|
if not valid_identifier(item.get("memoryId"), "mem-"):
|
|
reasons.append(f"selfPatch {field}[{index}].memoryId must start with 'mem-' and avoid whitespace")
|
|
check_patch_text(item.get("summary"), f"selfPatch {field}[{index}].summary", reasons, 600)
|
|
check_patch_unit(item.get("salience"), f"selfPatch {field}[{index}].salience", reasons)
|
|
check_patch_unit(item.get("confidence"), f"selfPatch {field}[{index}].confidence", reasons)
|
|
return len(value)
|
|
|
|
|
|
def review_goal_patch_array(patch: dict[str, Any], reasons: list[str]) -> int:
|
|
if "goals" not in patch:
|
|
return 0
|
|
value = patch["goals"]
|
|
if not isinstance(value, list):
|
|
reasons.append("selfPatch goals must be an array")
|
|
return 0
|
|
if len(value) > 6:
|
|
reasons.append("selfPatch goals may contain at most 6 records")
|
|
for index, item in enumerate(value):
|
|
if not isinstance(item, dict):
|
|
reasons.append(f"selfPatch goals[{index}] must be an object")
|
|
continue
|
|
if not valid_identifier(item.get("goalId"), "goal-"):
|
|
reasons.append(f"selfPatch goals[{index}].goalId must start with 'goal-' and avoid whitespace")
|
|
check_patch_text(item.get("description"), f"selfPatch goals[{index}].description", reasons, 700)
|
|
if item.get("scope") not in {"immediate", "scene", "case", "arc", "life"}:
|
|
reasons.append(f"selfPatch goals[{index}].scope is not a Ghostlight scope")
|
|
check_patch_unit(item.get("priority"), f"selfPatch goals[{index}].priority", reasons)
|
|
check_patch_text(item.get("emotionalStake"), f"selfPatch goals[{index}].emotionalStake", reasons, 400)
|
|
if item.get("status") not in {"active", "blocked", "dormant", "resolved", "abandoned"}:
|
|
reasons.append(f"selfPatch goals[{index}].status is not a Ghostlight status")
|
|
return len(value)
|
|
|
|
|
|
def review_value_patch_array(patch: dict[str, Any], reasons: list[str]) -> int:
|
|
if "values" not in patch:
|
|
return 0
|
|
value = patch["values"]
|
|
if not isinstance(value, list):
|
|
reasons.append("selfPatch values must be an array")
|
|
return 0
|
|
if len(value) > 6:
|
|
reasons.append("selfPatch values may contain at most 6 records")
|
|
for index, item in enumerate(value):
|
|
if not isinstance(item, dict):
|
|
reasons.append(f"selfPatch values[{index}] must be an object")
|
|
continue
|
|
if not valid_identifier(item.get("valueId"), "value-"):
|
|
reasons.append(f"selfPatch values[{index}].valueId must start with 'value-' and avoid whitespace")
|
|
check_patch_text(item.get("label"), f"selfPatch values[{index}].label", reasons, 240)
|
|
check_patch_unit(item.get("priority"), f"selfPatch values[{index}].priority", reasons)
|
|
if not isinstance(item.get("unforgivableIfBetrayed"), bool):
|
|
reasons.append(f"selfPatch values[{index}].unforgivableIfBetrayed must be boolean")
|
|
return len(value)
|
|
|
|
|
|
def review_private_notes(patch: dict[str, Any], reasons: list[str]) -> int:
|
|
if "privateNotes" not in patch:
|
|
return 0
|
|
value = patch["privateNotes"]
|
|
if not isinstance(value, list):
|
|
reasons.append("selfPatch privateNotes must be an array")
|
|
return 0
|
|
if len(value) > 6:
|
|
reasons.append("selfPatch privateNotes may contain at most 6 records")
|
|
for index, item in enumerate(value):
|
|
check_patch_text(item, f"selfPatch privateNotes[{index}]", reasons, 600)
|
|
return len(value)
|
|
|
|
|
|
def review_string_array(
|
|
patch: dict[str, Any],
|
|
field: str,
|
|
reasons: list[str],
|
|
*,
|
|
max_items: int,
|
|
max_len: int,
|
|
) -> None:
|
|
if field not in patch:
|
|
return
|
|
value = patch[field]
|
|
if not isinstance(value, list):
|
|
reasons.append(f"selfPatch {field} must be an array")
|
|
return
|
|
if len(value) > max_items:
|
|
reasons.append(f"selfPatch {field} may contain at most {max_items} records")
|
|
for index, item in enumerate(value):
|
|
check_patch_text(item, f"selfPatch {field}[{index}]", reasons, max_len)
|
|
|
|
|
|
def check_patch_text(value: Any, path: str, reasons: list[str], max_len: int) -> None:
|
|
if not isinstance(value, str) or not value.strip() or len(value) > max_len:
|
|
reasons.append(f"{path} must be non-empty text under {max_len} characters")
|
|
|
|
|
|
def check_patch_unit(value: Any, path: str, reasons: list[str]) -> None:
|
|
if not isinstance(value, (int, float)) or isinstance(value, bool) or not 0 <= float(value) <= 1:
|
|
reasons.append(f"{path} must be between 0 and 1")
|
|
|
|
|
|
def normalized_memory(item: dict[str, Any]) -> dict[str, Any]:
|
|
record = {
|
|
"memory_id": item["memoryId"],
|
|
"summary": item["summary"],
|
|
"salience": item["salience"],
|
|
"confidence": item["confidence"],
|
|
}
|
|
if "linkedEventIds" in item:
|
|
record["linked_event_ids"] = item["linkedEventIds"]
|
|
if "linkedRelationshipId" in item:
|
|
record["linked_relationship_id"] = item["linkedRelationshipId"]
|
|
return record
|
|
|
|
|
|
def normalized_goal(item: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"goal_id": item["goalId"],
|
|
"description": item["description"],
|
|
"scope": item["scope"],
|
|
"priority": item["priority"],
|
|
"emotional_stake": item["emotionalStake"],
|
|
"blockers": item.get("blockers", []),
|
|
"status": item["status"],
|
|
}
|
|
|
|
|
|
def normalized_value(item: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"value_id": item["valueId"],
|
|
"label": item["label"],
|
|
"priority": item["priority"],
|
|
"unforgivable_if_betrayed": item["unforgivableIfBetrayed"],
|
|
}
|
|
|
|
|
|
def upsert_by_id(records: list[dict[str, Any]], incoming: list[dict[str, Any]], id_field: str) -> list[dict[str, Any]]:
|
|
index = {record.get(id_field): record for record in records if isinstance(record, dict)}
|
|
for item in incoming:
|
|
index[item[id_field]] = item
|
|
return list(index.values())
|
|
|
|
|
|
def apply_self_patch(role_id: str, patch: dict[str, Any], *, agent_dir: Path = DEFAULT_AGENT_DIR) -> dict[str, Any]:
|
|
review = review_patch(role_id, patch)
|
|
if review["status"] != "accepted":
|
|
return review
|
|
_, filename = ROLE_TARGETS[role_id]
|
|
path = agent_dir / filename
|
|
document = load_json(path)
|
|
agent = document["agents"][0]
|
|
memories = agent["memories"]
|
|
memories["semantic"] = upsert_by_id(
|
|
memories["semantic"],
|
|
[normalized_memory(item) for item in patch.get("semanticMemories", [])],
|
|
"memory_id",
|
|
)
|
|
memories["episodic"] = upsert_by_id(
|
|
memories["episodic"],
|
|
[normalized_memory(item) for item in patch.get("episodicMemories", [])],
|
|
"memory_id",
|
|
)
|
|
memories["relationship_summaries"] = upsert_by_id(
|
|
memories["relationship_summaries"],
|
|
[normalized_memory(item) for item in patch.get("relationshipMemories", [])],
|
|
"memory_id",
|
|
)
|
|
agent["goals"] = upsert_by_id(
|
|
agent["goals"],
|
|
[normalized_goal(item) for item in patch.get("goals", [])],
|
|
"goal_id",
|
|
)
|
|
canonical = agent["canonical_state"]
|
|
canonical["values"] = upsert_by_id(
|
|
canonical["values"],
|
|
[normalized_value(item) for item in patch.get("values", [])],
|
|
"value_id",
|
|
)
|
|
private_notes = agent["identity"].setdefault("private_notes", [])
|
|
private_notes.extend(patch.get("privateNotes", []))
|
|
agent["identity"]["private_notes"] = private_notes[-32:]
|
|
write_json(path, document)
|
|
review["applied"] = True
|
|
return review
|
|
|
|
|
|
def validate_all(agent_dir: Path) -> list[str]:
|
|
errors: list[str] = []
|
|
for role_id, (_, filename) in ROLE_TARGETS.items():
|
|
path = agent_dir / filename
|
|
if not path.exists():
|
|
errors.append(f"{role_id}: missing {path}")
|
|
continue
|
|
errors.extend(validate_agent_file(path))
|
|
return errors
|
|
|
|
|
|
def parse_patch_arg(value: str) -> Any:
|
|
path = Path(value)
|
|
if path.exists():
|
|
return load_json(path)
|
|
return json.loads(value)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description="Review and apply Epiphany specialist self-memory patches.")
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
validate_parser = subparsers.add_parser("validate")
|
|
validate_parser.add_argument("--agent-dir", type=Path, default=DEFAULT_AGENT_DIR)
|
|
|
|
review_parser = subparsers.add_parser("review-patch")
|
|
review_parser.add_argument("--role-id", required=True, choices=sorted(ROLE_TARGETS))
|
|
review_parser.add_argument("--patch", required=True, help="JSON string or path to a JSON patch")
|
|
|
|
apply_parser = subparsers.add_parser("apply-patch")
|
|
apply_parser.add_argument("--role-id", required=True, choices=sorted(ROLE_TARGETS))
|
|
apply_parser.add_argument("--patch", required=True, help="JSON string or path to a JSON patch")
|
|
apply_parser.add_argument("--agent-dir", type=Path, default=DEFAULT_AGENT_DIR)
|
|
|
|
smoke_parser = subparsers.add_parser("smoke")
|
|
smoke_parser.add_argument("--agent-dir", type=Path, default=DEFAULT_AGENT_DIR)
|
|
|
|
args = parser.parse_args(argv)
|
|
if args.command == "validate":
|
|
errors = validate_all(args.agent_dir)
|
|
print(json.dumps({"ok": not errors, "errors": errors}, indent=2))
|
|
return 0 if not errors else 1
|
|
if args.command == "review-patch":
|
|
print(json.dumps(review_patch(args.role_id, parse_patch_arg(args.patch)), indent=2))
|
|
return 0
|
|
if args.command == "apply-patch":
|
|
patch = parse_patch_arg(args.patch)
|
|
result = apply_self_patch(args.role_id, patch, agent_dir=args.agent_dir)
|
|
print(json.dumps(result, indent=2))
|
|
return 0 if result["status"] == "accepted" else 1
|
|
if args.command == "smoke":
|
|
errors = validate_all(args.agent_dir)
|
|
if errors:
|
|
print(json.dumps({"ok": False, "phase": "validate", "errors": errors}, indent=2))
|
|
return 1
|
|
accepted_patch = {
|
|
"agentId": "epiphany.body",
|
|
"reason": "The Body should remember accepted graph growth must be source-grounded and bounded.",
|
|
"semanticMemories": [
|
|
{
|
|
"memoryId": "mem-body-smoke-source-grounding",
|
|
"summary": "A modeling self-memory request is acceptable when it improves future graph/checkpoint judgment without smuggling project truth.",
|
|
"salience": 0.74,
|
|
"confidence": 0.86,
|
|
}
|
|
],
|
|
}
|
|
accepted = review_patch("modeling", accepted_patch)
|
|
wrong_role = review_patch("verification", accepted_patch)
|
|
forbidden = review_patch(
|
|
"modeling",
|
|
{
|
|
"agentId": "epiphany.body",
|
|
"reason": "This tries to put project state in lane memory, which should be refused.",
|
|
"graphs": {},
|
|
"semanticMemories": [
|
|
{
|
|
"memoryId": "mem-body-bad-project-truth",
|
|
"summary": "Bad patch.",
|
|
"salience": 0.5,
|
|
"confidence": 0.5,
|
|
}
|
|
],
|
|
},
|
|
)
|
|
with TemporaryDirectory() as tmp:
|
|
tmp_dir = Path(tmp)
|
|
shutil.copytree(args.agent_dir, tmp_dir / "agents")
|
|
applied = apply_self_patch("modeling", accepted_patch, agent_dir=tmp_dir / "agents")
|
|
temp_errors = validate_all(tmp_dir / "agents")
|
|
ok = (
|
|
accepted["status"] == "accepted"
|
|
and wrong_role["status"] == "rejected"
|
|
and forbidden["status"] == "rejected"
|
|
and applied["status"] == "accepted"
|
|
and not temp_errors
|
|
)
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"ok": ok,
|
|
"accepted": accepted,
|
|
"wrongRole": wrong_role,
|
|
"forbidden": forbidden,
|
|
"applied": applied,
|
|
"tempValidationErrors": temp_errors,
|
|
},
|
|
indent=2,
|
|
)
|
|
)
|
|
return 0 if ok else 1
|
|
raise AssertionError(args.command)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|