"""Declarative agent definitions (harness alignment, v1). Zero third-party dependencies; standard library only. A definition is a versioned JSON document that bundles the fields a task already understands -- model, effort, approval, executor, plan -- plus a human-written instruction block and a list of required skills. It is the Foundry-style "agent defined declaratively, harness runs the loop" idea, cut down to what this codebase can actually honour today, with every boundary written down instead of implied: * Resolution happens at task creation (``POST /tasks`` with an inline ``agent`` object). Per field, an explicit body field wins over the definition; a field absent from both stays NULL and follows the existing worker-default chain. A definition that sets ``model`` therefore makes the task explicit for rule routing v0, and a definition that leaves ``model`` empty leaves routing to the worker. * Instructions reach the model only as a marked header prepended to the stored prompt (``[Agent: ]``), because the dispatch path (``cloud/sandbox.py`` -> ``host.py dispatch``) has no developerInstructions channel today. This is NOT claimed to be ``developerInstructions``: it is visible in the stored prompt, in the creation response, and in every later prompt read. Instructions are rejected with goal tasks and coordinator tasks, where the prompt is a goal or is never run as a model turn, so the header cannot silently pollute a goal text or a coordinator parent. * Skills are validated against the ``cloud/skills/`` pack at parse time (an unknown skill is a 400, never a silent drop). Seeding itself stays worker-global (``--seed-skills`` seeds the whole pack into the tenant codex-home); v1 does NOT claim per-agent seeding. When seeding is off, a listed skill is recorded in the creation response but is not guaranteed present at run time -- that is the documented v1 boundary, not a bug to be discovered later. * Routing policy is deliberately NOT part of a definition in v1: routing stays a worker-global policy (see ``cloud/models.py``). Putting a routing object inside a definition would either be silently dropped (a lie) or need a per-task policy column that does not exist. A definition that contains ``routing`` / ``routing_policy`` is rejected, so a typo or a v2 idea cannot masquerade as v1 behaviour. * Unknown top-level or defaults keys are rejected for the same reason: a declarative file must fail on typos, not silently ignore them. Definition identity (name/version/skills) is echoed in the creation response and, via the prompt header, in the stored prompt; v1 adds NO new tasks-table column, so ``GET /tasks`` shows the resolved fields and the header, not a separate ``agent`` column. That column is v2, not something v1 pretends to have. """ import json import os import re from pathlib import Path MAX_INSTRUCTIONS_CHARS = 8000 MAX_DESCRIPTION_CHARS = 1000 MAX_SKILLS = 20 NAME_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") EFFORTS = ("low", "medium", "high", "xhigh", "max") APPROVALS = ("ask", "auto", "full") EXECUTORS = ("local", "docker", "cloud") _TOP_LEVEL_KEYS = ("name", "version", "description", "instructions", "skills", "defaults") _DEFAULTS_KEYS = ("model", "effort", "approval", "executor", "plan") SKILLS_DIR = Path(__file__).resolve().parent / "skills" def list_known_skills(skills_dir=None): """Sorted skill names present in the pack (dirs with SKILL.md).""" base = Path(skills_dir) if skills_dir else SKILLS_DIR names = [] try: entries = sorted(base.iterdir()) except OSError: return names for entry in entries: if entry.is_dir() and (entry / "SKILL.md").is_file(): names.append(entry.name) return names def _clean_str(value, field): if not isinstance(value, str) or not value.strip(): raise ValueError("%s must be a non-empty string" % field) return value.strip() class AgentDefinition: """A parsed, validated agent definition.""" def __init__(self, name, version=None, description=None, instructions=None, skills=None, defaults=None): self.name = name self.version = version self.description = description self.instructions = instructions self.skills = list(skills or []) self.defaults = dict(defaults or {}) def to_summary(self): """The identity echoed in the task-creation response.""" return { "name": self.name, "version": self.version, "skills": list(self.skills), } def parse_agent_definition(data, known_skills=None): """Parse and validate one definition, or raise ValueError. ``known_skills`` is a list of pack skill names; when it is not None, every listed skill must be in it. Passing None skips the existence check (pure-function tests), it never invents a skill. """ if not isinstance(data, dict): raise ValueError("agent must be an object") unknown = sorted(set(data) - set(_TOP_LEVEL_KEYS)) if unknown: raise ValueError("agent has unknown field(s): %s" % ", ".join(unknown)) name = data.get("name") if not isinstance(name, str) or not NAME_RE.match(name): raise ValueError( "agent.name must match [a-z0-9][a-z0-9_-]{0,63}") version = None if data.get("version") is not None: version = _clean_str(data.get("version"), "agent.version") if len(version) > 64: raise ValueError("agent.version is too long") description = None if data.get("description") is not None: if not isinstance(data.get("description"), str): raise ValueError("agent.description must be a string") description = data["description"] if len(description) > MAX_DESCRIPTION_CHARS: raise ValueError("agent.description is too long") instructions = None if data.get("instructions") is not None: if not isinstance(data.get("instructions"), str): raise ValueError("agent.instructions must be a string") instructions = data["instructions"].strip() if not instructions: instructions = None elif len(instructions) > MAX_INSTRUCTIONS_CHARS: raise ValueError( "agent.instructions exceeds %d characters" % MAX_INSTRUCTIONS_CHARS) skills = [] if data.get("skills") is not None: raw_skills = data.get("skills") if not isinstance(raw_skills, list): raise ValueError("agent.skills must be a list") if len(raw_skills) > MAX_SKILLS: raise ValueError("agent.skills has too many entries") for skill in raw_skills: skill = _clean_str(skill, "agent.skills entry") if skill not in skills: skills.append(skill) if known_skills is not None: known = set(known_skills) missing = [s for s in skills if s not in known] if missing: raise ValueError("agent.skills unknown skill(s): %s" % ", ".join(missing)) defaults = {} if data.get("defaults") is not None: raw_defaults = data.get("defaults") if not isinstance(raw_defaults, dict): raise ValueError("agent.defaults must be an object") unknown_defaults = sorted(set(raw_defaults) - set(_DEFAULTS_KEYS)) if unknown_defaults: raise ValueError("agent.defaults has unknown field(s): %s" % ", ".join(unknown_defaults)) if raw_defaults.get("model") is not None: defaults["model"] = _clean_str(raw_defaults["model"], "agent.defaults.model") if raw_defaults.get("effort") is not None: if raw_defaults["effort"] not in EFFORTS: raise ValueError( "agent.defaults.effort must be low, medium, high," " xhigh or max") defaults["effort"] = raw_defaults["effort"] if raw_defaults.get("approval") is not None: if raw_defaults["approval"] not in APPROVALS: raise ValueError( "agent.defaults.approval must be ask, auto or full") defaults["approval"] = raw_defaults["approval"] if raw_defaults.get("executor") is not None: if raw_defaults["executor"] not in EXECUTORS: raise ValueError( "agent.defaults.executor must be local, docker" " or cloud") defaults["executor"] = raw_defaults["executor"] if raw_defaults.get("plan") is not None: plan = raw_defaults["plan"] if plan not in (True, False, 0, 1): raise ValueError("agent.defaults.plan must be true/false") defaults["plan"] = 1 if plan in (True, 1) else 0 if defaults.get("plan") and defaults.get("approval") == "full": raise ValueError( "plan 模式与 approval=full 冲突（只读 vs 完全访问）") return AgentDefinition(name=name, version=version, description=description, instructions=instructions, skills=skills, defaults=defaults) def load_agent_definition(path, agents_dir=None): """Load one definition from a JSON file (operator/local use only). This is NOT an API path: the server never loads a definition from a caller-supplied path (that would be a file-read primitive). When ``agents_dir`` is given, the resolved path must stay under it, so ``../`` cannot escape the definitions directory. """ candidate = Path(path) if agents_dir is not None and not candidate.is_absolute(): candidate = Path(agents_dir) / candidate resolved = candidate.resolve() if agents_dir is not None: base = Path(agents_dir).resolve() if base not in resolved.parents and resolved != base: raise ValueError("agent definition path escapes agents dir") try: with open(resolved, encoding="utf-8") as f: data = json.load(f) except OSError as exc: raise ValueError("cannot read agent definition: %s" % exc) except ValueError as exc: raise ValueError("agent definition is not valid JSON: %s" % exc) return parse_agent_definition(data) def resolve_task_fields(definition, body): """Resolve task fields: explicit body field wins, else definition. ``body`` is the task-creation body dict. Returns a dict with ``model``, ``effort``, ``approval``, ``executor`` (None when absent from both) and ``plan`` (int 0/1). It never mutates either input, and it never validates the resolved values -- the caller (server task creation) runs its existing validation on them, so a bad definition value is caught by the same 400 path as a bad body value. """ body = body if isinstance(body, dict) else {} defaults = definition.defaults if definition else {} resolved = {} for field in ("model", "effort", "approval", "executor"): value = body.get(field) if value is None: value = defaults.get(field) resolved[field] = value if "plan" in body: resolved["plan"] = 1 if body.get("plan") in (True, 1) else 0 elif defaults.get("plan") is not None: resolved["plan"] = defaults["plan"] else: resolved["plan"] = 0 return resolved def format_instructions_header(definition, prompt): """Prepend a definition's instructions to a prompt, marked. Returns the original prompt unchanged when there is no definition or no instructions, so callers that do not use definitions pay nothing and see no header. """ if definition is None or not definition.instructions: return prompt header = "[Agent: %s" % definition.name if definition.version: header += " %s" % definition.version header += "]" return "%s\n%s\n\n%s" % (header, definition.instructions, prompt)