415669c2e4
Some checks are pending
CI / lint-and-test (push) Waiting to run
* Preserve candidate reasoning and disable automatic chat reinforcement * Retain exact-row guards for combined claim audits
114 lines
3.8 KiB
Python
114 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Disable automatic chat-to-memory and chat-to-skill reinforcement for leoclean."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
POLICY_VERSION = "teleo-leoclean-learning-policy-v1"
|
|
BACKUP_SUFFIX = ".before-teleo-learning-policy"
|
|
MEMORY_NUDGE_INTERVAL = 0
|
|
SKILL_CREATION_NUDGE_INTERVAL = 0
|
|
|
|
|
|
def _load_config(path: Path) -> dict[str, Any]:
|
|
if path.is_symlink():
|
|
raise ValueError("refusing symlinked profile config")
|
|
raw = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
|
|
if not isinstance(raw, dict):
|
|
raise ValueError("profile config must be a YAML mapping")
|
|
return raw
|
|
|
|
|
|
def policy_projection(config: dict[str, Any]) -> dict[str, int | None]:
|
|
memory = config.get("memory") if isinstance(config.get("memory"), dict) else {}
|
|
skills = config.get("skills") if isinstance(config.get("skills"), dict) else {}
|
|
return {
|
|
"memory_nudge_interval": memory.get("nudge_interval"),
|
|
"skill_creation_nudge_interval": skills.get("creation_nudge_interval"),
|
|
}
|
|
|
|
|
|
def enforce_policy(path: Path, *, check: bool = False) -> tuple[dict[str, Any], int]:
|
|
config = _load_config(path)
|
|
before = policy_projection(config)
|
|
expected = {
|
|
"memory_nudge_interval": MEMORY_NUDGE_INTERVAL,
|
|
"skill_creation_nudge_interval": SKILL_CREATION_NUDGE_INTERVAL,
|
|
}
|
|
changed = before != expected
|
|
|
|
if check:
|
|
return {
|
|
"policy_version": POLICY_VERSION,
|
|
"status": "change_required" if changed else "compliant",
|
|
"target": str(path),
|
|
"policy": before,
|
|
}, 1 if changed else 0
|
|
|
|
backup_path = path.with_name(path.name + BACKUP_SUFFIX)
|
|
if changed:
|
|
memory = config.setdefault("memory", {})
|
|
skills = config.setdefault("skills", {})
|
|
if not isinstance(memory, dict) or not isinstance(skills, dict):
|
|
raise ValueError("memory and skills config sections must be YAML mappings")
|
|
memory["nudge_interval"] = MEMORY_NUDGE_INTERVAL
|
|
skills["creation_nudge_interval"] = SKILL_CREATION_NUDGE_INTERVAL
|
|
|
|
if not backup_path.exists():
|
|
shutil.copy2(path, backup_path)
|
|
|
|
mode = path.stat().st_mode & 0o777
|
|
rendered = yaml.safe_dump(config, sort_keys=False, allow_unicode=False)
|
|
with tempfile.NamedTemporaryFile(
|
|
"w", encoding="utf-8", dir=path.parent, prefix=f".{path.name}.", delete=False
|
|
) as handle:
|
|
handle.write(rendered)
|
|
temp_path = Path(handle.name)
|
|
try:
|
|
temp_path.chmod(mode)
|
|
os.replace(temp_path, path)
|
|
finally:
|
|
temp_path.unlink(missing_ok=True)
|
|
|
|
after = policy_projection(_load_config(path))
|
|
if after != expected:
|
|
raise ValueError("profile config did not retain the required learning policy")
|
|
return {
|
|
"policy_version": POLICY_VERSION,
|
|
"status": "updated" if changed else "already_compliant",
|
|
"target": str(path),
|
|
"backup": str(backup_path) if changed else None,
|
|
"policy": after,
|
|
}, 0
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("target", type=Path)
|
|
parser.add_argument("--check", action="store_true")
|
|
args = parser.parse_args()
|
|
try:
|
|
result, code = enforce_policy(args.target, check=args.check)
|
|
except (OSError, ValueError, yaml.YAMLError) as exc:
|
|
result = {
|
|
"policy_version": POLICY_VERSION,
|
|
"status": "error",
|
|
"target": str(args.target),
|
|
"error": type(exc).__name__,
|
|
}
|
|
code = 2
|
|
print(json.dumps(result, sort_keys=True))
|
|
return code
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|