"""Bounded requirement-to-skill authoring for Aureon's CodeArchitect. The model-facing surface accepts one strict JSON plan. Model output never becomes source code: this module validates an allowlisted declarative plan, renders deterministic Python, performs compile/static checks, rehearses the skill against Aureon's in-memory VM backend, and only then commits a ``VALIDATED`` skill that still requires explicit approval. """ from __future__ import annotations import copy import hashlib import json import os import re import tempfile from pathlib import Path from typing import Any, Dict, Mapping, Tuple from aureon.code_architect.executor import SkillExecutor from aureon.code_architect.skill import Skill, SkillLevel, SkillProposal, SkillStatus from aureon.code_architect.skill_library import SkillLibrary, get_skill_library from aureon.code_architect.validator import SkillValidator MAX_REQUIREMENT_CHARS = 4_000 MAX_PLAN_CHARS = 32_000 MAX_STEPS = 64 _NAME_RE = re.compile(r"^[a-z][a-z0-9_]{2,63}$") _INPUT_RE = re.compile(r"^[a-z][a-z0-9_]{0,47}$") _SECRET_RE = re.compile( r"(?:BEGIN (?:RSA |OPENSSH |EC )?PRIVATE KEY|\bsk_(?:live|proj)_[A-Za-z0-9_-]{8,}|" r"\b(?:password|api[_-]?secret|private[_-]?key|credential|access[_-]?token)\b)", re.IGNORECASE, ) # Shell/PowerShell are intentionally absent. Requirement-authored skills may # only compose bounded human-interface and read-only observation primitives. PRIMITIVE_PARAM_KINDS: Dict[str, Dict[str, str]] = { "screenshot": {}, "mouse_move": {"x": "coordinate", "y": "coordinate"}, "left_click": {"x": "optional_coordinate", "y": "optional_coordinate"}, "right_click": {"x": "optional_coordinate", "y": "optional_coordinate"}, "middle_click": {"x": "optional_coordinate", "y": "optional_coordinate"}, "double_click": {"x": "optional_coordinate", "y": "optional_coordinate"}, "triple_click": {"x": "optional_coordinate", "y": "optional_coordinate"}, "left_click_drag": { "start_x": "coordinate", "start_y": "coordinate", "end_x": "coordinate", "end_y": "coordinate", }, "scroll": { "x": "coordinate", "y": "coordinate", "direction": "direction", "amount": "scroll_amount", }, "type_text": {"text": "text"}, "press_key": {"key": "key"}, "hotkey": {"keys": "keys"}, "get_cursor_position": {}, "get_screen_size": {}, "list_windows": {}, "get_active_window": {}, "focus_window": {"title": "title"}, "wait": {"seconds": "wait_seconds"}, } REQUIRED_PARAMS: Dict[str, frozenset[str]] = { "mouse_move": frozenset({"x", "y"}), "left_click_drag": frozenset({"start_x", "start_y", "end_x", "end_y"}), "scroll": frozenset({"x", "y"}), "type_text": frozenset({"text"}), "press_key": frozenset({"key"}), "hotkey": frozenset({"keys"}), "focus_window": frozenset({"title"}), } class RequirementSkillError(ValueError): """Raised internally when a requirement plan violates its bounded schema.""" def _canonical_json(value: Any) -> str: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False) def _sha256_text(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest() def _is_input_ref(value: Any) -> bool: return isinstance(value, dict) and set(value) == {"input", "default"} def _validate_literal(kind: str, value: Any, *, context: str) -> None: if kind in {"coordinate", "optional_coordinate"}: if value is None and kind == "optional_coordinate": return if isinstance(value, bool) or not isinstance(value, (int, float)): raise RequirementSkillError(f"{context}_must_be_numeric") if not 0 <= float(value) <= 16_384: raise RequirementSkillError(f"{context}_out_of_range") return if kind == "direction": if value not in {"up", "down", "left", "right"}: raise RequirementSkillError(f"{context}_direction_invalid") return if kind == "scroll_amount": if isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= 20: raise RequirementSkillError(f"{context}_scroll_amount_invalid") return if kind == "wait_seconds": if isinstance(value, bool) or not isinstance(value, (int, float)) or not 0 <= float(value) <= 1: raise RequirementSkillError(f"{context}_wait_invalid") return if kind == "text": if not isinstance(value, str) or len(value) > 2_000: raise RequirementSkillError(f"{context}_text_invalid") if _SECRET_RE.search(value): raise RequirementSkillError(f"{context}_secret_like_text_blocked") return if kind in {"key", "title"}: limit = 64 if kind == "key" else 240 if not isinstance(value, str) or not value.strip() or len(value) > limit: raise RequirementSkillError(f"{context}_{kind}_invalid") if _SECRET_RE.search(value): raise RequirementSkillError(f"{context}_secret_like_text_blocked") return if kind == "keys": if ( not isinstance(value, list) or not 1 <= len(value) <= 5 or any(not isinstance(item, str) or not item.strip() or len(item) > 32 for item in value) ): raise RequirementSkillError(f"{context}_keys_invalid") return raise RequirementSkillError(f"{context}_unknown_param_kind") def _python_value(value: Any) -> str: if _is_input_ref(value): input_name = str(value["input"]) return f"kwargs.get({input_name!r}, {value['default']!r})" return repr(value) class RequirementSkillBuilder: """Create staged, approval-required skills from strict declarative plans.""" def __init__( self, *, library: SkillLibrary | None = None, validator: SkillValidator | None = None, adapter: Any = None, ) -> None: self.library = library if library is not None else get_skill_library() self.validator = validator or SkillValidator(strict_static=True) self.adapter = adapter @staticmethod def planning_system_prompt() -> str: primitives = sorted(PRIMITIVE_PARAM_KINDS) return ( "Return exactly one JSON object and no Markdown. The only keys are name, description, steps, " "and optional sample_inputs. Each step has exactly primitive and params. Primitive must be one " f"of {json.dumps(primitives)}. Never emit Python, shell, PowerShell, credentials, secrets, " "network calls, or file operations. A reusable parameter may be encoded only as " '{"input":"field_name","default":}. ' "Keep the plan bounded and deterministic." ) def _plan_with_adapter(self, requirement: str, adapter: Any) -> str: response = adapter.prompt( messages=[{"role": "user", "content": f"Requirement:\n{requirement}"}], system=self.planning_system_prompt(), max_tokens=2_048, temperature=0.0, ) return str(getattr(response, "text", response) or "") def _parse_and_validate_plan( self, raw_plan: str | Mapping[str, Any], ) -> Tuple[Dict[str, Any], Dict[str, Tuple[str, Any]]]: if isinstance(raw_plan, Mapping): text = _canonical_json(dict(raw_plan)) else: text = str(raw_plan or "").strip() if not text or len(text) > MAX_PLAN_CHARS: raise RequirementSkillError("plan_size_invalid") if _SECRET_RE.search(text): raise RequirementSkillError("plan_secret_scan_failed") try: plan = json.loads(text) except (TypeError, ValueError) as exc: raise RequirementSkillError("strict_json_plan_required") from exc if not isinstance(plan, dict): raise RequirementSkillError("plan_object_required") allowed_top = {"name", "description", "steps", "sample_inputs"} if set(plan) - allowed_top or not {"name", "description", "steps"}.issubset(plan): raise RequirementSkillError("plan_keys_invalid") name = plan.get("name") description = plan.get("description") steps = plan.get("steps") if not isinstance(name, str) or not _NAME_RE.fullmatch(name): raise RequirementSkillError("skill_name_invalid") if not isinstance(description, str) or not description.strip() or len(description) > 240: raise RequirementSkillError("skill_description_invalid") if not isinstance(steps, list) or not 1 <= len(steps) <= MAX_STEPS: raise RequirementSkillError("skill_steps_invalid") input_specs: Dict[str, Tuple[str, Any]] = {} normalized_steps = [] for index, raw_step in enumerate(steps): if not isinstance(raw_step, dict) or set(raw_step) != {"primitive", "params"}: raise RequirementSkillError(f"step_{index}_shape_invalid") primitive = raw_step.get("primitive") params = raw_step.get("params") if primitive not in PRIMITIVE_PARAM_KINDS: raise RequirementSkillError(f"step_{index}_primitive_not_allowlisted") if not isinstance(params, dict): raise RequirementSkillError(f"step_{index}_params_invalid") kinds = PRIMITIVE_PARAM_KINDS[str(primitive)] if set(params) - set(kinds): raise RequirementSkillError(f"step_{index}_param_not_allowlisted") if not REQUIRED_PARAMS.get(str(primitive), frozenset()).issubset(params): raise RequirementSkillError(f"step_{index}_required_param_missing") if primitive in { "left_click", "right_click", "middle_click", "double_click", "triple_click", } and (("x" in params) != ("y" in params)): raise RequirementSkillError(f"step_{index}_click_coordinates_incomplete") normalized_params: Dict[str, Any] = {} for param_name, value in params.items(): kind = kinds[param_name] if _is_input_ref(value): input_name = value.get("input") default = value.get("default") if not isinstance(input_name, str) or not _INPUT_RE.fullmatch(input_name): raise RequirementSkillError(f"step_{index}_{param_name}_input_name_invalid") if _SECRET_RE.search(input_name): raise RequirementSkillError(f"step_{index}_{param_name}_secret_input_blocked") _validate_literal(kind, default, context=f"step_{index}_{param_name}_default") prior = input_specs.get(input_name) if prior is not None and prior != (kind, default): raise RequirementSkillError(f"input_{input_name}_definition_conflict") input_specs[input_name] = (kind, default) normalized_params[param_name] = {"input": input_name, "default": default} else: if isinstance(value, dict): raise RequirementSkillError(f"step_{index}_{param_name}_object_invalid") _validate_literal(kind, value, context=f"step_{index}_{param_name}") normalized_params[param_name] = value normalized_steps.append({"primitive": primitive, "params": normalized_params}) samples = plan.get("sample_inputs") or {} if not isinstance(samples, dict) or set(samples) - set(input_specs): raise RequirementSkillError("sample_inputs_invalid") normalized_samples: Dict[str, Any] = {} for input_name, value in samples.items(): kind, _default = input_specs[input_name] _validate_literal(kind, value, context=f"sample_{input_name}") normalized_samples[input_name] = value normalized = { "name": name, "description": description.strip(), "steps": normalized_steps, "sample_inputs": normalized_samples, } return normalized, input_specs @staticmethod def _render_source(plan: Mapping[str, Any]) -> str: lines = [f"def {plan['name']}(**kwargs):", f" {plan['description']!r}", " results = []"] for step in plan["steps"]: args = ["session_id=kwargs.get('session_id')"] for key, value in step["params"].items(): args.append(f"{key}={_python_value(value)}") lines.append(f" results.append(vm_{step['primitive']}({', '.join(args)}))") lines.extend( [ " ok = True", " for result in results:", " if not bool(result.get('ok', False)):", " ok = False", " return {'ok': ok, 'steps': len(results), 'results': results}", ] ) return "\n".join(lines) + "\n" @staticmethod def _params_schema(input_specs: Mapping[str, Tuple[str, Any]]) -> Dict[str, Any]: properties: Dict[str, Any] = {"session_id": {"type": "string"}} for name, (kind, default) in input_specs.items(): if kind in {"coordinate", "optional_coordinate", "scroll_amount", "wait_seconds"}: json_type = "number" elif kind == "keys": json_type = "array" else: json_type = "string" properties[name] = {"type": json_type, "default": default} return {"type": "object", "properties": properties, "additionalProperties": False} def _validate_in_staging( self, proposal: SkillProposal, simulation_inputs: Mapping[str, Any], ) -> Dict[str, Any]: validation: Dict[str, Any] = { "compile_ok": False, "static_safe": False, "simulation_ok": False, "simulation_backend": "simulated", } try: compile(proposal.code, f"", "exec") validation["compile_ok"] = True except Exception as exc: validation["compile_error"] = f"{type(exc).__name__}: {exc}" return validation static_ok, static_errors = self.validator.static_check(proposal.code) validation["static_safe"] = bool(static_ok) validation["static_errors"] = list(static_errors[:10]) if not static_ok: return validation from aureon.autonomous.vm_control import VMControlDispatcher dispatcher = VMControlDispatcher() session_id = "" try: session_id = dispatcher.create_session( backend="simulated", name=f"requirement-skill-{proposal.name}", make_default=True, ) controller = dispatcher.get_session(session_id) if controller is None: validation["simulation_error"] = "simulated_session_missing" return validation controller.arm(dry_run=False) with tempfile.TemporaryDirectory(prefix="aureon_requirement_skill_stage_") as temp_dir: stage_library = SkillLibrary(storage_dir=Path(temp_dir) / "skills") staged_skill = Skill.from_proposal(proposal) staged_skill.status = SkillStatus.APPROVED staged_skill.tags = ["requirement_generated", "live_execution_disabled"] stage_library.add(staged_skill) executor = SkillExecutor( library=stage_library, dispatcher=dispatcher, validator=self.validator, ) params = dict(simulation_inputs) params["session_id"] = session_id result = executor.execute(staged_skill, params=params) returned = result.return_value validation["simulation_ok"] = bool( result.ok and isinstance(returned, dict) and returned.get("ok") is True ) validation["simulation_error"] = str(result.error or "") if result.ok and not validation["simulation_ok"]: validation["simulation_error"] = "simulated_skill_reported_failure" validation["simulation_duration_s"] = round(float(result.duration_s), 6) except Exception as exc: validation["simulation_error"] = f"{type(exc).__name__}: {exc}" finally: dispatcher.destroy_all() return validation def _restore_library_file(self, existed: bool, content: bytes) -> None: path = self.library.library_path if not existed: path.unlink(missing_ok=True) return path.parent.mkdir(parents=True, exist_ok=True) fd, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".rollback", dir=str(path.parent)) try: with os.fdopen(fd, "wb") as handle: handle.write(content) handle.flush() os.fsync(handle.fileno()) os.replace(temp_name, path) finally: Path(temp_name).unlink(missing_ok=True) def _persist_skill_transaction(self, skill: Skill) -> Tuple[bool, bool, str]: lock = getattr(self.library, "_lock", None) if lock is None: raise RequirementSkillError("skill_library_lock_unavailable") with lock: prior = copy.deepcopy(self.library.get(skill.name)) existed = self.library.library_path.exists() before = self.library.library_path.read_bytes() if existed else b"" try: self.library.add(skill, persist=False) self.library.save() payload = json.loads(self.library.library_path.read_text(encoding="utf-8")) records = payload.get("skills", []) if isinstance(payload, dict) else [] committed = next( (item for item in records if isinstance(item, dict) and item.get("name") == skill.name), None, ) if committed is None or _canonical_json(committed) != _canonical_json(skill.to_dict()): raise OSError("skill_library_commit_verification_failed") return True, False, "" except Exception as exc: rollback_error = "" try: self.library.remove(skill.name, persist=False) if prior is not None: self.library.add(prior, persist=False) self._restore_library_file(existed, before) except Exception as rollback_exc: rollback_error = f"; rollback={type(rollback_exc).__name__}: {rollback_exc}" return False, not rollback_error, f"{type(exc).__name__}: {exc}{rollback_error}" def _remove_skill_transaction(self, name: str) -> Tuple[bool, bool, str]: lock = getattr(self.library, "_lock", None) if lock is None: raise RequirementSkillError("skill_library_lock_unavailable") with lock: prior = copy.deepcopy(self.library.get(name)) if prior is None: return False, False, "skill_not_found" existed = self.library.library_path.exists() before = self.library.library_path.read_bytes() if existed else b"" try: self.library.remove(name, persist=False) self.library.save() payload = json.loads(self.library.library_path.read_text(encoding="utf-8")) records = payload.get("skills", []) if isinstance(payload, dict) else [] if any(isinstance(item, dict) and item.get("name") == name for item in records): raise OSError("skill_library_removal_verification_failed") return True, False, "" except Exception as exc: rollback_error = "" try: self.library.add(prior, persist=False) self._restore_library_file(existed, before) except Exception as rollback_exc: rollback_error = f"; rollback={type(rollback_exc).__name__}: {rollback_exc}" return False, not rollback_error, f"{type(exc).__name__}: {exc}{rollback_error}" def build( self, requirement: str, *, plan: str | Mapping[str, Any] | None = None, adapter: Any = None, simulation_inputs: Mapping[str, Any] | None = None, ) -> Dict[str, Any]: requirement_text = str(requirement or "").strip() if not requirement_text or len(requirement_text) > MAX_REQUIREMENT_CHARS: return {"ok": False, "status": "rejected", "reason": "requirement_size_invalid"} if _SECRET_RE.search(requirement_text): return {"ok": False, "status": "rejected", "reason": "requirement_secret_scan_failed"} try: raw_plan = plan if raw_plan is None: planner = adapter or self.adapter if planner is None: raise RequirementSkillError("declarative_plan_or_adapter_required") raw_plan = self._plan_with_adapter(requirement_text, planner) normalized, input_specs = self._parse_and_validate_plan(raw_plan) if self.library.contains(normalized["name"]): raise RequirementSkillError("skill_name_already_exists") samples = dict(normalized.get("sample_inputs") or {}) for input_name, value in dict(simulation_inputs or {}).items(): if input_name not in input_specs: raise RequirementSkillError("simulation_input_not_declared") kind, _default = input_specs[input_name] _validate_literal(kind, value, context=f"simulation_{input_name}") samples[input_name] = value source = self._render_source(normalized) plan_digest = _sha256_text(_canonical_json(normalized)) proposal = SkillProposal( name=normalized["name"], description=normalized["description"], level=SkillLevel.TASK, category="requirement_generated", code=source, entry_function=normalized["name"], params_schema=self._params_schema(input_specs), dependencies=[], observation_sources=[f"requirement_sha256:{_sha256_text(requirement_text)}"], created_by="requirement_skill_builder", reasoning="Deterministically rendered from an allowlisted declarative primitive plan.", target="vm", ) validation = self._validate_in_staging(proposal, samples) if not all(validation.get(key) is True for key in ("compile_ok", "static_safe", "simulation_ok")): return { "ok": False, "status": "validation_failed", "reason": "compile_static_or_simulation_failed", "skill_name": proposal.name, "plan_digest": plan_digest, "source_digest": _sha256_text(source), "validation": validation, "persisted": False, "rolled_back": True, } skill = Skill.from_proposal(proposal) skill.status = SkillStatus.VALIDATED skill.tags = [ "requirement_generated", "requires_explicit_approval", "live_execution_disabled", f"plan_sha256:{plan_digest}", ] committed, rolled_back, error = self._persist_skill_transaction(skill) if not committed: return { "ok": False, "status": "library_commit_failed", "reason": error, "skill_name": skill.name, "plan_digest": plan_digest, "source_digest": _sha256_text(source), "validation": validation, "persisted": False, "rolled_back": rolled_back, } return { "ok": True, "status": "validated_pending_approval", "skill_name": skill.name, "skill_status": skill.status.value, "plan_digest": plan_digest, "source_digest": _sha256_text(source), "step_count": len(normalized["steps"]), "validation": validation, "persisted": True, "live_execution_enabled": False, "library_path": str(self.library.library_path), } except RequirementSkillError as exc: return {"ok": False, "status": "rejected", "reason": str(exc), "persisted": False} except Exception as exc: return { "ok": False, "status": "error", "reason": f"{type(exc).__name__}: {exc}", "persisted": False, } def approve_skill( self, name: str, *, reviewer: str, enable_live: bool = False, ) -> Dict[str, Any]: skill = self.library.get(str(name or "")) if skill is None: return {"ok": False, "status": "not_found", "reason": "skill_not_found"} if "requirement_generated" not in skill.tags: return {"ok": False, "status": "rejected", "reason": "not_requirement_generated"} reviewer_text = str(reviewer or "").strip() if not reviewer_text or len(reviewer_text) > 120: return {"ok": False, "status": "rejected", "reason": "reviewer_required"} approved = copy.deepcopy(skill) approved.status = SkillStatus.APPROVED approved.tags = [tag for tag in approved.tags if not tag.startswith("approved_by:")] approved.tags.append(f"approved_by:{reviewer_text}") if enable_live: approved.tags = [tag for tag in approved.tags if tag != "live_execution_disabled"] elif "live_execution_disabled" not in approved.tags: approved.tags.append("live_execution_disabled") committed, rolled_back, error = self._persist_skill_transaction(approved) return { "ok": committed, "status": "approved" if committed else "approval_commit_failed", "skill_name": approved.name, "skill_status": approved.status.value if committed else skill.status.value, "live_execution_enabled": committed and "live_execution_disabled" not in approved.tags, "rolled_back": rolled_back, "reason": error, } def reject_skill(self, name: str, *, reviewer: str, reason: str = "rejected") -> Dict[str, Any]: skill = self.library.get(str(name or "")) if skill is None: return {"ok": False, "status": "not_found", "reason": "skill_not_found"} if "requirement_generated" not in skill.tags: return {"ok": False, "status": "rejected", "reason": "not_requirement_generated"} if not str(reviewer or "").strip(): return {"ok": False, "status": "rejected", "reason": "reviewer_required"} removed, rolled_back, error = self._remove_skill_transaction(skill.name) return { "ok": removed, "status": "removed_rejected_skill" if removed else "rejection_commit_failed", "skill_name": skill.name, "reviewer": str(reviewer).strip()[:120], "reject_reason": str(reason or "rejected")[:240], "rolled_back": rolled_back, "reason": error, } __all__ = [ "MAX_STEPS", "PRIMITIVE_PARAM_KINDS", "RequirementSkillBuilder", "RequirementSkillError", ]