Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """Read-only, fail-closed Spynel recognition; no dispatch or model imports at import time.""" | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import os | |
| import re | |
| import sys | |
| from pathlib import Path | |
| from urllib.parse import urlparse | |
| ROOT = Path(__file__).resolve().parents[1] | |
| ALIASES = { | |
| "coding agent": "developer", | |
| "repo manager": "repository-manager", | |
| "prompt improver": "prompt-improver", | |
| "gepa specialist": "prompt-improver", | |
| } | |
| TOKEN = re.compile(r"(?<!\S)([@/])([A-Za-z0-9][A-Za-z0-9_-]{0,63})(?=\s|$)") | |
| # Mask examples without changing offsets in the original message. | |
| EXCLUDED = re.compile( | |
| r'```[\s\S]*?(?:```|$)|~~~[\s\S]*?(?:~~~|$)|`[^`\n]*`|"[^"\n]*"|\'[^\'\n]*\'|(?:[a-z][a-z0-9+.-]*://|www\.)\S+|(?:^|\n)\s*>[^\n]*' | |
| ) | |
| INTENT = re.compile( | |
| r"(?:^|[.!?\n]\s*|\b(?:and|or)\s+)(?:please\s+)?(?:ask|have|let|tell|use|delegate\s+to)\s+(?:the\s+)?", | |
| re.IGNORECASE, | |
| ) | |
| def visible(text): | |
| return EXCLUDED.sub(lambda m: " " * len(m.group()), text) | |
| def normalize(value): | |
| return re.sub(r"[-_\s]+", " ", value).strip().casefold() | |
| def discover(project=None): | |
| if str(ROOT) not in sys.path: | |
| sys.path.insert(0, str(ROOT)) | |
| from helpers import subagents | |
| profiles = { | |
| key: item.title | |
| for key, item in subagents.get_available_agents_dict(project).items() | |
| } | |
| external = {} | |
| for key, value in os.environ.items(): | |
| match = re.fullmatch(r"SPYNEL_(SDK|AGENT)_([A-Z0-9_]+)_URL", key) | |
| if not match or not value: | |
| continue | |
| parsed = urlparse(value) | |
| if ( | |
| parsed.scheme == "https" | |
| and parsed.hostname | |
| and not parsed.username | |
| and not parsed.password | |
| ): | |
| kind, name = match.groups() | |
| external[ | |
| ( | |
| "sdk" if kind == "SDK" else "external_agent", | |
| name.lower().replace("_", "-"), | |
| ) | |
| ] = name | |
| return profiles, external | |
| def workspace_candidates(text, root): | |
| root = Path(root).resolve() | |
| found = set() | |
| # Require path-oriented language or a delegation request; plain mentions do | |
| # not establish workspace intent. Never scan/create user directories. | |
| if not ( | |
| INTENT.search(text) | |
| or re.search( | |
| r"\b(?:work on|folder|directory|workspace|inspect)\b", text, re.IGNORECASE | |
| ) | |
| ): | |
| return [] | |
| for word in re.findall(r"(?<![\w./-])[\w.-]+(?:/[\w.-]+)*(?![\w./-])", text): | |
| word = word.rstrip(".") if word not in {".", ".."} else word | |
| try: | |
| candidate = (root / word).resolve() | |
| if ( | |
| candidate != root | |
| and candidate.is_relative_to(root) | |
| and candidate.is_dir() | |
| ): | |
| found.add(str(candidate)) | |
| except (OSError, RuntimeError): | |
| continue | |
| return sorted(found) | |
| def resolve( | |
| text, | |
| profiles, | |
| external=None, | |
| workspace_root="/a0/usr/workdir", | |
| *, | |
| match_workspace=True, | |
| ): | |
| """Recognize a candidate; only prepare_route validates it for execution.""" | |
| external = external or {} | |
| clean = visible(text) | |
| workspaces = workspace_candidates(clean, workspace_root) if match_workspace else [] | |
| base = { | |
| "decision": "local", | |
| "target_type": None, | |
| "target_id": None, | |
| "workspace_path": workspaces[0] if len(workspaces) == 1 else None, | |
| "recognition": "deterministic", | |
| "ambiguous": len(workspaces) > 1, | |
| "message": text, | |
| "workspace_candidates": workspaces, | |
| } | |
| aliases = {} | |
| def add(label, target): | |
| aliases.setdefault(normalize(label), set()).add(target) | |
| for name, title in profiles.items(): | |
| add(name, ("internal_agent", name)) | |
| add(title, ("internal_agent", name)) | |
| for alias, name in ALIASES.items(): | |
| if name in profiles: | |
| add(alias, ("internal_agent", name)) | |
| for target, title in external.items(): | |
| # An environment-registered agent overrides its internal counterpart. | |
| if target[0] == "external_agent": | |
| for values in aliases.values(): | |
| if ("internal_agent", target[1]) in values: | |
| values.discard(("internal_agent", target[1])) | |
| values.add(target) | |
| add(target[1], target) | |
| add(title, target) | |
| explicit = list(TOKEN.finditer(clean)) | |
| targets = set() | |
| if explicit: | |
| # Preserve historical @profile + /sdk combination: SDK selects gateway. | |
| selected = [m for m in explicit if m[1] == "/"] or explicit | |
| for match in selected: | |
| name = match[2].lower() | |
| target = ( | |
| ("sdk", name) | |
| if match[1] == "/" | |
| else ("external_agent", name) | |
| if ("external_agent", name) in external | |
| else ("internal_agent", name) | |
| ) | |
| if ( | |
| target[0] == "internal_agent" | |
| and name not in profiles | |
| or target[0] != "internal_agent" | |
| and target not in external | |
| ): | |
| return {**base, "decision": "abstain", "reason": "unknown_target"} | |
| targets.add(target) | |
| else: | |
| for intent in INTENT.finditer(clean): | |
| rest = clean[intent.end() :] | |
| matches = [] | |
| for label, values in aliases.items(): | |
| pattern = r"[-_\s]+".join(re.escape(word) for word in label.split()) | |
| match = re.match(pattern + r"(?=$|[\s,.!?;:])", rest, re.IGNORECASE) | |
| if match: | |
| matches.append((match.end(), values)) | |
| if matches: | |
| longest = max(length for length, _ in matches) | |
| for length, values in matches: | |
| if length == longest: | |
| tail = rest[length:] | |
| if re.match(r"\.\w", tail): | |
| continue | |
| if re.match(r"\s+(?:and|or)\s+", tail, re.IGNORECASE): | |
| return { | |
| **base, | |
| "decision": "abstain", | |
| "ambiguous": True, | |
| "reason": "multiple_targets", | |
| } | |
| targets.update(values) | |
| else: | |
| return {**base, "decision": "abstain", "reason": "unknown_target"} | |
| if len(targets) > 1 or len(workspaces) > 1: | |
| return {**base, "decision": "abstain", "ambiguous": True, "reason": "ambiguous"} | |
| if targets: | |
| kind, name = targets.pop() | |
| return {**base, "decision": "dispatch", "target_type": kind, "target_id": name} | |
| return base | |
| def load_workspace_aliases(): | |
| """Optional trusted JSON configuration; user/model text cannot define aliases.""" | |
| value = json.loads(os.getenv("SPYNEL_WORKSPACE_ALIASES", "{}")) | |
| if not isinstance(value, dict) or any( | |
| not isinstance(k, str) | |
| or not k.strip() | |
| or not isinstance(v, str) | |
| or not v.strip() | |
| for k, v in value.items() | |
| ): | |
| raise ValueError("SPYNEL_WORKSPACE_ALIASES must map names to existing paths") | |
| normalized = {normalize(k): v for k, v in value.items()} | |
| if len(normalized) != len(value): | |
| raise ValueError("Ambiguous workspace aliases") | |
| return normalized | |
| def _target_spans(text, profiles, external): | |
| spans = [m.span() for m in TOKEN.finditer(text)] | |
| labels = [ | |
| *profiles, | |
| *profiles.values(), | |
| *(key[1] for key in external), | |
| *external.values(), | |
| *ALIASES, | |
| ] | |
| for intent in INTENT.finditer(text): | |
| for label in labels: | |
| pattern = r"[-_\s]+".join( | |
| re.escape(part) for part in normalize(label).split() | |
| ) | |
| match = re.match( | |
| pattern + r"(?=$|[\s,.!?;:])", text[intent.end() :], re.IGNORECASE | |
| ) | |
| if match: | |
| spans.append((intent.end(), intent.end() + match.end())) | |
| return spans | |
| def workspace_context(text, root, *, aliases=None, selected=None, target_spans=()): | |
| """Validate referenced directories AND files, retaining no unverified path.""" | |
| root = Path(root).resolve() | |
| aliases = aliases or {} | |
| clean = visible(text) | |
| directories, files, errors = set(), set(), [] | |
| def add(raw, required): | |
| raw = raw.rstrip(".,;:!?") | |
| if not raw or raw in {".", ".."}: | |
| if required: | |
| errors.append("workspace_invalid") | |
| return | |
| value = aliases.get(normalize(raw), raw) | |
| required = required or (root / value).is_symlink() | |
| try: | |
| candidate = (root / value).resolve(strict=True) | |
| if not candidate.is_relative_to(root): | |
| if required or (root / value).exists(): | |
| errors.append("workspace_escape") | |
| return | |
| if candidate.is_dir(): | |
| if candidate != root or required: | |
| directories.add(str(candidate)) | |
| elif candidate.is_file(): | |
| files.add(str(candidate)) | |
| elif required: | |
| errors.append("workspace_invalid") | |
| except (OSError, RuntimeError, ValueError): | |
| if required: | |
| errors.append("workspace_missing") | |
| if selected is not None: | |
| add(selected, True) | |
| used = list(target_spans) | |
| # Multiword aliases are configured, never guessed by a model. | |
| for alias in sorted(aliases, key=len, reverse=True): | |
| pattern = ( | |
| r"(?<![\w/-])" | |
| + r"[-_\s]+".join(re.escape(p) for p in alias.split()) | |
| + r"(?![\w/-])" | |
| ) | |
| for match in re.finditer(pattern, clean, re.IGNORECASE): | |
| if not any( | |
| match.start() < end and match.end() > start for start, end in used | |
| ): | |
| add(alias, True) | |
| used.append(match.span()) | |
| for match in re.finditer( | |
| r"(?<![\w./-])(?:/[\w./-]+|[\w.-]+(?:/[\w.-]+)*)(?![\w./-])", clean | |
| ): | |
| if any(match.start() < end and match.end() > start for start, end in used): | |
| continue | |
| word = match.group().rstrip(".,;:!?") | |
| before, after = clean[: match.start()], clean[match.end() :] | |
| required = ( | |
| "/" in word | |
| or bool(re.search(r"\.[A-Za-z0-9]{1,8}$", word)) | |
| or bool( | |
| re.search( | |
| r"\b(?:file|folder|directory|workspace)\s+(?:the\s+)?$", | |
| before, | |
| re.IGNORECASE, | |
| ) | |
| ) | |
| or ( | |
| bool(re.match(r"\s+(?:file|folder|directory)\b", after, re.IGNORECASE)) | |
| and word.casefold() | |
| not in {"inspect", "check", "fix", "work", "on", "use", "in"} | |
| ) | |
| or ( | |
| "-" in word | |
| and bool( | |
| re.search( | |
| r"\b(?:inspect|check|fix|work on)\s+$", before, re.IGNORECASE | |
| ) | |
| ) | |
| ) | |
| ) | |
| # Syntax words themselves are not path references. | |
| if word.casefold() not in {"file", "folder", "directory", "workspace", "the"}: | |
| add(word, required) | |
| # Quoted paths are allowed only after a visible path marker, not within | |
| # quoted examples or fenced source code. | |
| for match in re.finditer( | |
| r"""\b(?:file|folder|directory|workspace)\s+["']([^"'\n]+)["']""", | |
| text, | |
| re.IGNORECASE, | |
| ): | |
| if clean[match.start()] != " ": | |
| add(match[1], True) | |
| path = None | |
| if len(directories) > 1: | |
| errors.append("workspace_ambiguous") | |
| elif directories: | |
| path = next(iter(directories)) | |
| if any(not Path(file).is_relative_to(path) for file in files): | |
| errors.append("workspace_ambiguous") | |
| elif files: | |
| parents = {str(Path(file).parent) for file in files} | |
| if len(parents) == 1: | |
| path = parents.pop() | |
| else: | |
| errors.append("workspace_ambiguous") | |
| return { | |
| "workspace_path": path if not errors else None, | |
| "workspace_files": sorted(files) if not errors else [], | |
| "workspace_candidates": sorted(directories), | |
| "workspace_error": errors[0] if errors else None, | |
| } | |
| def prepare_route( | |
| text, | |
| profiles, | |
| external=None, | |
| workspace_root="/a0/usr/workdir", | |
| *, | |
| profile=None, | |
| sdk=None, | |
| workspace=None, | |
| workspace_aliases=None, | |
| ): | |
| """Final deterministic validation for recognized or main-model-selected work. | |
| No model call is made. Explicit tokens beat selections; a selected identifier | |
| cannot override an unknown or ambiguous natural-language destination. | |
| """ | |
| external = external or {} | |
| result = resolve(text, profiles, external, workspace_root, match_workspace=False) | |
| clean = visible(text) | |
| explicit = list(TOKEN.finditer(clean)) | |
| explicit_profiles = {m[2].lower() for m in explicit if m[1] == "@"} | |
| explicit_sdks = {m[2].lower() for m in explicit if m[1] == "/"} | |
| if len(explicit_profiles) > 1 or len(explicit_sdks) > 1: | |
| result = { | |
| **result, | |
| "decision": "abstain", | |
| "reason": "ambiguous", | |
| "ambiguous": True, | |
| } | |
| # Validate all identifiers, even @profile accompanying a /sdk gateway. | |
| for name in explicit_profiles: | |
| if name not in profiles and ("external_agent", name) not in external: | |
| return {**result, "decision": "abstain", "reason": "unknown_profile"} | |
| for name in explicit_sdks: | |
| if ("sdk", name) not in external: | |
| return {**result, "decision": "abstain", "reason": "unknown_sdk"} | |
| if not explicit and result["decision"] == "local" and (profile or sdk): | |
| # Ordinary interpretation happens in Spynel's EXISTING main interaction. | |
| # This API only validates its proposed identifiers; it never classifies. | |
| if not clean.strip(): | |
| return {**result, "decision": "abstain", "reason": "no_instruction"} | |
| if ( | |
| profile | |
| and profile not in profiles | |
| and ("external_agent", profile) not in external | |
| ): | |
| return {**result, "decision": "abstain", "reason": "unknown_profile"} | |
| if sdk and ("sdk", sdk) not in external: | |
| return {**result, "decision": "abstain", "reason": "unknown_sdk"} | |
| kind = ( | |
| "sdk" | |
| if sdk | |
| else "external_agent" | |
| if ("external_agent", profile) in external | |
| else "internal_agent" | |
| ) | |
| result = { | |
| **result, | |
| "decision": "dispatch", | |
| "target_type": kind, | |
| "target_id": sdk or profile, | |
| "recognition": "validated_selection", | |
| } | |
| if result["decision"] == "abstain" and result.get("reason") != "ambiguous": | |
| return result | |
| paths = workspace_context( | |
| text, | |
| workspace_root, | |
| aliases=workspace_aliases, | |
| selected=workspace, | |
| target_spans=_target_spans(clean, profiles, external), | |
| ) | |
| result = {**result, **paths} | |
| if paths["workspace_error"]: | |
| return { | |
| **result, | |
| "decision": "abstain", | |
| "reason": paths["workspace_error"], | |
| "ambiguous": paths["workspace_error"] == "workspace_ambiguous", | |
| } | |
| result["explicit_profile"] = next(iter(explicit_profiles), None) | |
| result["agent_profile"] = result["explicit_profile"] or ( | |
| profile if result.get("recognition") == "validated_selection" else None | |
| ) | |
| return result | |
| def profile_override_conflicts(agent): | |
| """Read-only audit of active higher-priority Spynel instructions and skills. | |
| Metadata-only title/avatar changes and byte-equivalent copies are allowed. | |
| Divergent routing instructions block use until the operator reconciles them. | |
| Never delete, migrate or print local prompt contents. | |
| """ | |
| if str(ROOT) not in sys.path: | |
| sys.path.insert(0, str(ROOT)) | |
| from helpers import subagents | |
| bundled = ROOT / "agents/spynel" | |
| conflicts = [] | |
| for location in subagents.get_paths( | |
| agent, include_default=False, include_user=False | |
| ): | |
| location = Path(location) | |
| if location.resolve() == bundled.resolve(): | |
| break | |
| candidates = [ | |
| *location.glob("prompts/agent.system.main*.md"), | |
| *location.glob("skills/spynel-dispatch/**/*"), | |
| *location.glob("extensions/**/*.py"), | |
| ] | |
| for path in candidates: | |
| if not path.is_file() or "__pycache__" in path.parts: | |
| continue | |
| counterpart = bundled / path.relative_to(location) | |
| try: | |
| if ( | |
| not counterpart.is_file() | |
| or path.read_bytes() != counterpart.read_bytes() | |
| ): | |
| conflicts.append(str(path)) | |
| except OSError: | |
| conflicts.append(str(path)) | |
| for name in ("agent.yaml", "agent.json", "_context.md"): | |
| path = location / name | |
| if not path.is_file(): | |
| continue | |
| try: | |
| if name == "_context.md": | |
| context = path.read_text().strip() | |
| elif name.endswith("json"): | |
| context = json.loads(path.read_text()).get("context", "") | |
| else: | |
| from helpers import yaml as yaml_helper | |
| context = (yaml_helper.loads(path.read_text()) or {}).get( | |
| "context", "" | |
| ) | |
| if context: | |
| conflicts.append(str(path)) | |
| except (OSError, ValueError, AttributeError): | |
| conflicts.append(str(path)) | |
| return sorted(set(conflicts)) | |
| def profile_conflicts(project=None): | |
| from types import SimpleNamespace | |
| agent = SimpleNamespace( | |
| config=SimpleNamespace(profile="spynel"), | |
| context=SimpleNamespace(get_data=lambda key: project), | |
| ) | |
| return profile_override_conflicts(agent) | |
| def main(): | |
| parser = argparse.ArgumentParser(description=__doc__) | |
| mode = parser.add_mutually_exclusive_group(required=True) | |
| mode.add_argument("--resolve-only", action="store_true") | |
| mode.add_argument("--check-profile", action="store_true") | |
| parser.add_argument("--profile") | |
| parser.add_argument("--sdk") | |
| parser.add_argument("--workspace") | |
| parser.add_argument("--workspace-root", default="/a0/usr/workdir") | |
| parser.add_argument("--project") | |
| parser.add_argument("message", nargs="?", default="") | |
| args = parser.parse_args() | |
| if args.check_profile: | |
| conflicts = profile_conflicts(args.project) | |
| print( | |
| json.dumps( | |
| {"status": "blocked" if conflicts else "ok", "conflicts": conflicts} | |
| ) | |
| ) | |
| return 2 if conflicts else 0 | |
| profiles, external = discover(args.project) | |
| print( | |
| json.dumps( | |
| prepare_route( | |
| args.message, | |
| profiles, | |
| external, | |
| args.workspace_root, | |
| profile=args.profile, | |
| sdk=args.sdk, | |
| workspace=args.workspace, | |
| workspace_aliases=load_workspace_aliases(), | |
| ), | |
| ensure_ascii=False, | |
| ) | |
| ) | |
| if __name__ == "__main__": | |
| raise SystemExit(main()) | |