cortex-code — Skillopedia

Cortex Code Integration Skill Install This skill enables your coding agent to leverage Cortex Code's specialized Snowflake expertise by intelligently routing Snowflake-related operations to Cortex Code CLI in headless mode. Architecture Overview Routing Principle : ONLY Snowflake operations → Cortex Code. Everything else → your coding agent. Key Components : - Dynamic skill discovery at session initialization - LLM-based semantic routing (not keyword matching) - Security wrapper with approval modes (prompt/auto/envelope only) - Stateless Cortex execution with context enrichment - Hybrid memor…

, line.strip()):\n continue\n\n # New format: \" - skill-name: /path/to/skill\"\n new_format_match = re.match(r'^\\s*-\\s+(\\S+?):\\s+', line)\n if new_format_match:\n skill_name = new_format_match.group(1).strip()\n else:\n # Old format: \"skill-name /path/to/skill\"\n parts = line.split()\n if not parts:\n continue\n skill_name = parts[0].strip(':').strip()\n\n # Read the skill's SKILL.md to get description and triggers\n skill_info = read_skill_metadata(skill_name)\n if skill_info:\n skills[skill_name] = skill_info\n\n return skills\n\n\ndef read_skill_metadata(skill_name):\n \"\"\"Read SKILL.md frontmatter for a specific skill.\"\"\"\n # Cortex bundled skills are typically in ~/.local/share/cortex/{version}/bundled_skills/\n cortex_share = Path.home() / \".local/share/cortex\"\n\n # Find the most recent version directory\n if not cortex_share.exists():\n return None\n\n version_dirs = sorted([d for d in cortex_share.iterdir() if d.is_dir()], reverse=True)\n\n for version_dir in version_dirs:\n bundled_skills = version_dir / \"bundled_skills\"\n if not bundled_skills.exists():\n continue\n\n # Look for skill directory\n skill_path = bundled_skills / skill_name / \"SKILL.md\"\n if skill_path.exists():\n return parse_skill_md(skill_path)\n\n return None\n\n\ndef parse_skill_md(skill_path):\n \"\"\"Parse SKILL.md file and extract frontmatter.\"\"\"\n try:\n with open(skill_path, 'r') as f:\n content = f.read()\n\n # Extract YAML frontmatter\n frontmatter_match = re.match(r'^---\\n(.*?)\\n---', content, re.DOTALL)\n if not frontmatter_match:\n return None\n\n frontmatter = frontmatter_match.group(1)\n\n # Simple YAML parsing for name and description\n name_match = re.search(r'name:\\s*(.+)', frontmatter)\n desc_match = re.search(r'description:\\s*[\"\\']?(.+?)[\"\\']?

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

, frontmatter, re.MULTILINE | re.DOTALL)\n\n if name_match and desc_match:\n name = name_match.group(1).strip().strip('\"\\'')\n description = desc_match.group(1).strip().strip('\"\\'')\n\n # Extract \"Use when\" trigger patterns from body\n triggers = extract_triggers(content)\n\n return {\n \"name\": name,\n \"description\": description,\n \"triggers\": triggers\n }\n except Exception as e:\n print(f\"Error parsing {skill_path}: {e}\", file=sys.stderr)\n return None\n\n\ndef extract_triggers(content):\n \"\"\"Extract trigger phrases from skill content.\"\"\"\n triggers = []\n\n # Look for \"Use when\", \"Trigger\", \"When to use\" sections\n trigger_patterns = [\n r'(?:Use when|When to use|Trigger).*?:\\s*(.+?)(?=\\n\\n|\\#\\#)',\n r'- Use (?:when|for|if):\\s*(.+?)

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

\n ]\n\n for pattern in trigger_patterns:\n matches = re.finditer(pattern, content, re.MULTILINE | re.DOTALL)\n for match in matches:\n trigger_text = match.group(1).strip()\n # Clean up and split by common separators\n phrases = re.split(r'[,;]|\\n-', trigger_text)\n triggers.extend([p.strip() for p in phrases if p.strip()])\n\n return triggers[:10] # Limit to 10 most relevant triggers\n\n\ndef main():\n \"\"\"Main discovery function.\"\"\"\n # Parse command line arguments\n parser = argparse.ArgumentParser(description=\"Discover Cortex Code capabilities\")\n parser.add_argument(\n \"--cache-dir\",\n type=Path,\n help=\"Cache directory for storing capabilities (default: from config or ~/.cache/cortex-skill)\"\n )\n args = parser.parse_args()\n\n # Determine cache directory\n if args.cache_dir:\n cache_dir = args.cache_dir\n else:\n # Get default from config\n config_manager = ConfigManager()\n cache_dir_str = config_manager.get(\"security.cache_dir\")\n cache_dir = Path(cache_dir_str).expanduser()\n\n # Discover capabilities\n capabilities = discover_cortex_skills()\n\n # Cache using CacheManager with SHA256 fingerprint validation\n try:\n cache_manager = CacheManager(cache_dir)\n cache_manager.write(\"cortex-capabilities\", capabilities, ttl=86400) # 24-hour TTL\n print(f\"Discovered {len(capabilities)} Cortex skills\", file=sys.stderr)\n print(f\"Cached to: {cache_dir / 'cortex-capabilities.json'}\", file=sys.stderr)\n except Exception as e:\n # If cache fails, log warning but continue\n print(f\"Warning: Failed to cache capabilities: {e}\", file=sys.stderr)\n print(f\"Discovered {len(capabilities)} Cortex skills\", file=sys.stderr)\n\n # Output the capabilities\n print(json.dumps(capabilities, indent=2))\n\n return 0\n\n\nif __name__ == \"__main__\":\n sys.exit(main())\n","content_type":"text/x-python; charset=utf-8","language":"python","size":6464,"content_sha256":"d2066177cc7868d3fb8aa5819228a3691c94f656a5cd4f8ccee3eef62afa186c"},{"filename":"scripts/execute_cortex.py","content":"#!/usr/bin/env python3\n\"\"\"\nExecutes Cortex Code in headless mode with streaming output parsing.\nUses --output-format stream-json for streaming results.\nHandles tool use events and final results.\n\"\"\"\n\nimport json\nimport os\nimport subprocess\nimport sys\nimport argparse\nimport threading\nimport queue\nimport time\nfrom pathlib import Path\nfrom typing import List, Dict, Optional\n\ntry:\n from security.prompt_sanitizer import PromptSanitizer\nexcept Exception:\n PromptSanitizer = None\n\n\n# Known tools for inversion logic (allowed -> disallowed)\nKNOWN_TOOLS = [\n \"Read\", \"Write\", \"Edit\", \"Bash\", \"Grep\", \"Glob\",\n \"snowflake_sql_execute\", \"data_diff\", \"snowflake_query\"\n]\n\nDESTRUCTIVE_SHELL_TOOLS = [\n \"Bash\",\n \"Bash(rm *)\", \"Bash(rm -rf *)\", \"Bash(rm -r *)\",\n \"Bash(sudo *)\", \"Bash(chmod 777 *)\",\n \"Bash(git push *)\", \"Bash(git reset --hard *)\"\n]\n\nREAD_ONLY_TOOLS = [\"Edit\", \"Write\", \"Bash\"] + DESTRUCTIVE_SHELL_TOOLS\nUNKNOWN_TOOL_SENTINEL = \"*\"\n\n\ndef _redact_error_output(error_text: str) -> str:\n \"\"\"Redact sensitive data before returning/logging error output.\"\"\"\n if PromptSanitizer is None:\n return error_text\n return PromptSanitizer().sanitize(error_text)\n\n\ndef invert_tools_to_disallowed(allowed_tools: List[str]) -> List[str]:\n \"\"\"\n Convert allowed tools list to disallowed tools list.\n\n For prompt mode: when security wrapper predicts/approves specific tools,\n we need to invert the list to block all OTHER tools via --disallowed-tools.\n\n Args:\n allowed_tools: List of tool names that ARE allowed\n\n Returns:\n List of tool names that should be disallowed (inverse of allowed)\n\n Example:\n allowed = [\"Read\", \"Grep\"]\n disallowed = [\"Write\", \"Edit\", \"Bash\", \"Glob\", ...other tools...]\n \"\"\"\n inverted = [tool for tool in KNOWN_TOOLS if tool not in allowed_tools]\n inverted.append(UNKNOWN_TOOL_SENTINEL)\n return inverted\n\n\ndef execute_cortex_streaming(prompt: str, connection: Optional[str] = None,\n disallowed_tools: Optional[List[str]] = None,\n envelope: str = \"RW\",\n approval_mode: str = \"prompt\",\n allowed_tools: Optional[List[str]] = None,\n timeout_seconds: int = 300,\n deploy_confirmed: bool = False) -> Dict:\n \"\"\"\n Execute Cortex with streaming JSON output in programmatic mode.\n\n Uses --output-format stream-json for streaming results.\n Tools are controlled via --disallowed-tools blocklists for safety.\n\n Args:\n prompt: The enriched prompt to send to Cortex\n connection: Optional Snowflake connection name\n disallowed_tools: Optional list of tools to explicitly block\n envelope: Security envelope mode (RO, RW, RESEARCH, DEPLOY, NONE)\n approval_mode: Approval mode (prompt, auto, envelope_only)\n allowed_tools: Optional list of tools that ARE allowed (for prompt mode)\n\n Returns:\n Dictionary with execution results\n \"\"\"\n if approval_mode in [\"auto\", \"envelope_only\"] and envelope == \"NONE\":\n raise ValueError(\"NONE envelope is not allowed in auto or envelope_only approval modes\")\n if approval_mode in [\"auto\", \"envelope_only\"] and envelope == \"DEPLOY\" and not deploy_confirmed:\n raise ValueError(\"DEPLOY envelope requires explicit confirmation\")\n\n # Build command in print mode. The prompt is delivered with -p; do not add\n # --input-format stream-json here. Cortex treats that flag as JSON stdin\n # input mode, so combining it with -p and closed stdin can emit only the\n # initial session event and exit before the prompt is processed.\n cmd = [\n \"cortex\",\n \"-p\", prompt,\n \"--output-format\", \"stream-json\"\n ]\n\n # Add connection if specified\n if connection:\n cmd.extend([\"-c\", connection])\n\n # Step 1: Handle approval mode — build disallowed tools list for envelope security.\n # Do NOT use --allowed-tools: it creates a \"must match pattern\" check that\n # blocks Snowflake MCP tools.\n final_disallowed_tools = disallowed_tools or []\n\n if approval_mode == \"prompt\":\n # Prompt mode: invert allowed_tools to disallowed_tools\n # In prompt mode, we ONLY use allowed_tools (don't merge with envelope)\n if allowed_tools is not None:\n # User approved specific tools - block everything else\n inverted_tools = invert_tools_to_disallowed(allowed_tools)\n # Merge with existing disallowed tools (but NOT envelope tools)\n final_disallowed_tools = list(set(final_disallowed_tools) | set(inverted_tools))\n else:\n # No tools approved - block all known tools\n final_disallowed_tools = list(set(final_disallowed_tools) | set(KNOWN_TOOLS))\n\n elif approval_mode in [\"envelope_only\", \"auto\"]:\n # Envelope-only or auto mode: apply envelope-based security via blocklist.\n envelope_tools = []\n if envelope == \"RO\":\n # Read-only: block all write operations\n envelope_tools = READ_ONLY_TOOLS\n elif envelope in [\"RW\", \"DEPLOY\"]:\n # RW and DEPLOY may allow shell usage, but still block destructive\n # shell patterns by default. Explicit custom disallowed_tools can\n # add stricter policy on top.\n envelope_tools = DESTRUCTIVE_SHELL_TOOLS\n elif envelope == \"RESEARCH\":\n # Research: read-only plus web access\n envelope_tools = READ_ONLY_TOOLS\n # Merge envelope tools with final disallowed list\n if envelope_tools:\n final_disallowed_tools = list(set(final_disallowed_tools) | set(envelope_tools))\n\n # Step 3: Add final disallowed tools to command\n if final_disallowed_tools:\n for tool in final_disallowed_tools:\n cmd.extend([\"--disallowed-tools\", tool])\n\n debug_cmd = f\"cortex -p \\\"...\\\" --output-format stream-json\"\n if connection:\n debug_cmd += f\" -c {connection}\"\n if final_disallowed_tools:\n debug_cmd += f\" --disallowed-tools {' '.join(final_disallowed_tools[:3])}{'...' if len(final_disallowed_tools) > 3 else ''}\"\n print(debug_cmd, file=sys.stderr)\n\n process = None\n stderr_lines = []\n\n def _read_stderr(stderr):\n if stderr is None:\n return\n for stderr_line in stderr:\n stderr_lines.append(stderr_line)\n\n def _kill_process():\n if not process:\n return\n process.kill()\n try:\n process.wait(timeout=1)\n except Exception:\n pass\n\n try:\n # Start process. stdin=DEVNULL prevents accidental reads from the parent\n # terminal; prompt delivery is handled exclusively by -p print mode.\n process = subprocess.Popen(\n cmd,\n stdout=subprocess.PIPE,\n stderr=subprocess.PIPE,\n stdin=subprocess.DEVNULL,\n text=True,\n bufsize=1\n )\n\n stderr_thread = threading.Thread(target=_read_stderr, args=(process.stderr,), daemon=True)\n stderr_thread.start()\n\n results = {\n \"session_id\": None,\n \"events\": [],\n \"permission_requests\": [],\n \"final_result\": None,\n \"error\": None\n }\n\n stdout_queue = queue.Queue()\n stdout_errors = queue.Queue()\n\n def _read_stdout(stdout):\n if stdout is None:\n stdout_queue.put(None)\n return\n try:\n for stdout_line in stdout:\n stdout_queue.put(stdout_line)\n except Exception as exc:\n stdout_errors.put(exc)\n finally:\n stdout_queue.put(None)\n\n stdout_thread = threading.Thread(target=_read_stdout, args=(process.stdout,), daemon=True)\n stdout_thread.start()\n\n timed_out = False\n deadline = time.monotonic() + timeout_seconds\n while True:\n remaining = deadline - time.monotonic()\n if remaining \u003c= 0:\n timed_out = True\n break\n\n try:\n line = stdout_queue.get(timeout=remaining)\n except queue.Empty:\n timed_out = True\n break\n\n if line is None:\n if not stdout_errors.empty():\n raise stdout_errors.get()\n break\n\n if not line.strip():\n continue\n\n try:\n event = json.loads(line)\n results[\"events\"].append(event)\n\n event_type = event.get(\"type\")\n\n # Extract session ID\n if event_type == \"system\" and event.get(\"subtype\") == \"init\":\n results[\"session_id\"] = event.get(\"session_id\")\n print(f\"→ Started Cortex session: {results['session_id']}\", file=sys.stderr)\n\n # Handle assistant responses\n elif event_type == \"assistant\":\n message = event.get(\"message\", {})\n content = message.get(\"content\", [])\n\n for item in content:\n if item.get(\"type\") == \"text\":\n print(f\"[Cortex] {item.get('text', '')}\", file=sys.stderr)\n\n elif item.get(\"type\") == \"tool_use\":\n tool_name = item.get(\"name\")\n print(f\"[Cortex] Using tool: {tool_name}\", file=sys.stderr)\n\n # Handle permission requests (via user messages with tool_result containing denials)\n elif event_type == \"user\":\n message = event.get(\"message\", {})\n content = message.get(\"content\", [])\n\n for item in content:\n if item.get(\"type\") == \"tool_result\":\n tool_content = item.get(\"content\", \"\")\n tool_content_text = json.dumps(tool_content) if isinstance(tool_content, list) else str(tool_content)\n if \"Permission denied\" in tool_content_text or \"denied\" in tool_content_text.lower():\n results[\"permission_requests\"].append({\n \"tool_use_id\": item.get(\"tool_use_id\"),\n \"content\": tool_content\n })\n print(f\"[Cortex] Permission request detected: {tool_content_text}\", file=sys.stderr)\n\n # Handle final result\n elif event_type == \"result\":\n results[\"final_result\"] = event.get(\"result\")\n print(f\"[Cortex] Result: {event.get('result')}\", file=sys.stderr)\n\n except json.JSONDecodeError as e:\n print(f\"Warning: Failed to parse line: {line[:100]}... Error: {e}\", file=sys.stderr)\n continue\n\n if timed_out:\n raise subprocess.TimeoutExpired(cmd=cmd, timeout=timeout_seconds)\n\n # Wait for process to complete\n process.wait(timeout=timeout_seconds)\n stderr_thread.join(timeout=1)\n\n # Check for errors\n if process.returncode != 0:\n stderr_output = _redact_error_output(\"\".join(stderr_lines))\n results[\"error\"] = stderr_output\n print(f\"Error: Cortex exited with code {process.returncode}\", file=sys.stderr)\n print(f\"Stderr: {stderr_output}\", file=sys.stderr)\n\n return results\n\n except subprocess.TimeoutExpired:\n _kill_process()\n return {\n \"session_id\": None,\n \"events\": [],\n \"permission_requests\": [],\n \"final_result\": None,\n \"error\": f\"Cortex execution timed out after {timeout_seconds} seconds\"\n }\n\n except Exception as e:\n _kill_process()\n return {\n \"session_id\": None,\n \"events\": [],\n \"permission_requests\": [],\n \"final_result\": None,\n \"error\": _redact_error_output(str(e))\n }\n\n\ndef _resolve_output_path(output_file: str) -> Path:\n \"\"\"Resolve output path under a safe output directory.\"\"\"\n base_dir = Path(os.environ.get(\"CORTEX_CODE_OUTPUT_DIR\", Path.cwd())).expanduser().resolve()\n output_path = Path(output_file).expanduser()\n if not output_path.is_absolute():\n output_path = base_dir / output_path\n output_path = output_path.resolve()\n try:\n output_path.relative_to(base_dir)\n except ValueError as exc:\n raise ValueError(f\"Output file must be under {base_dir}\") from exc\n return output_path\n\n\ndef main():\n \"\"\"Main execution function.\"\"\"\n parser = argparse.ArgumentParser(description=\"Execute Cortex Code headlessly\")\n parser.add_argument(\"--prompt\", required=True, help=\"Prompt to send to Cortex\")\n parser.add_argument(\"--connection\", \"-c\", help=\"Snowflake connection name\")\n parser.add_argument(\"--disallowed-tools\", nargs=\"+\", help=\"Tools to explicitly block\")\n parser.add_argument(\"--envelope\", default=\"RW\",\n choices=[\"RO\", \"RW\", \"RESEARCH\", \"DEPLOY\", \"NONE\"],\n help=\"Security envelope mode (default: RW)\")\n parser.add_argument(\"--approval-mode\", default=\"prompt\",\n choices=[\"prompt\", \"auto\", \"envelope_only\"],\n help=\"Approval mode (default: prompt)\")\n parser.add_argument(\"--allowed-tools\", nargs=\"+\",\n help=\"Tools that are allowed (for prompt mode)\")\n parser.add_argument(\"--timeout\", type=int, default=300,\n help=\"Maximum seconds to wait for Cortex execution (default: 300)\")\n parser.add_argument(\"--deploy-confirmed\", action=\"store_true\",\n help=\"Required explicit confirmation for DEPLOY envelope in non-interactive modes\")\n parser.add_argument(\"--output-file\", help=\"Write JSON results to this file instead of stdout\")\n parser.add_argument(\"--stream\", action=\"store_true\", help=\"Stream output (always true)\")\n args = parser.parse_args()\n\n # Execute Cortex\n results = execute_cortex_streaming(\n args.prompt,\n connection=args.connection,\n disallowed_tools=args.disallowed_tools,\n envelope=args.envelope,\n approval_mode=args.approval_mode,\n allowed_tools=args.allowed_tools,\n timeout_seconds=args.timeout,\n deploy_confirmed=args.deploy_confirmed\n )\n\n # Output results as JSON\n output = json.dumps(results, indent=2)\n if args.output_file:\n try:\n output_path = _resolve_output_path(args.output_file)\n except ValueError as exc:\n print(json.dumps({\"error\": str(exc)}, indent=2))\n return 1\n output_path.parent.mkdir(parents=True, exist_ok=True)\n output_path.write_text(output + \"\\n\")\n else:\n print(output)\n\n # Exit with appropriate code\n if results.get(\"error\"):\n return 1\n\n return 0\n\n\nif __name__ == \"__main__\":\n sys.exit(main())\n","content_type":"text/x-python; charset=utf-8","language":"python","size":15106,"content_sha256":"89a0c8cd8a0c58aaddb606bd09d6ff02b9f90962fa1b05be56bccaf34ffd2156"},{"filename":"scripts/predict_tools.py","content":"#!/usr/bin/env python3\n\"\"\"\nPredicts which Cortex tools will be needed based on the user prompt and capabilities.\nEnhanced with confidence scoring for approval handler.\n\"\"\"\n\nimport json\nimport sys\nimport argparse\nfrom pathlib import Path\nfrom security.cache_manager import CacheManager\nfrom security.config_manager import ConfigManager\n\n\n# Tool prediction mappings with weighted patterns\nTOOL_PATTERNS = {\n \"snowflake_sql_execute\": [\n \"select\", \"insert\", \"update\", \"delete\", \"query\", \"sql\",\n \"table\", \"database\", \"data\", \"snowflake\"\n ],\n \"bash\": [\n \"run\", \"execute\", \"command\", \"script\", \"install\", \"shell\"\n ],\n \"read\": [\n \"read\", \"show\", \"display\", \"view\", \"check\", \"inspect\", \"examine\"\n ],\n \"write\": [\n \"create\", \"write\", \"generate\", \"save\", \"output\", \"file\"\n ],\n \"glob\": [\n \"find\", \"search\", \"list\", \"files\", \"directory\", \"locate\"\n ],\n \"grep\": [\n \"search\", \"find\", \"pattern\", \"match\", \"contains\"\n ]\n}\n\n\n# Always include these base tools for Snowflake operations\nBASE_SNOWFLAKE_TOOLS = [\"snowflake_sql_execute\", \"bash\", \"read\"]\n\n\ndef load_capabilities():\n \"\"\"Load cached Cortex capabilities through CacheManager.\"\"\"\n try:\n config_manager = ConfigManager()\n cache_dir = Path(config_manager.get(\"security.cache_dir\")).expanduser()\n cache_manager = CacheManager(cache_dir)\n return cache_manager.read(\"cortex-capabilities\") or {}\n except Exception as exc:\n print(f\"Warning: Failed to load Cortex capabilities from cache: {exc}\", file=sys.stderr)\n return {}\n\n\ndef predict_tools(prompt, envelope=None):\n \"\"\"\n Predict required tools based on prompt analysis with confidence scoring.\n\n Args:\n prompt: User prompt to analyze\n envelope: Optional envelope dict with capabilities\n\n Returns:\n dict with:\n - tools: list of predicted tool names\n - confidence: float 0-1 indicating prediction confidence\n - reasoning: str explaining the prediction\n \"\"\"\n prompt_lower = prompt.lower()\n predicted = set(BASE_SNOWFLAKE_TOOLS)\n matched_patterns = []\n\n # Check each tool pattern and track matches\n for tool, patterns in TOOL_PATTERNS.items():\n tool_matches = []\n for pattern in patterns:\n if pattern in prompt_lower:\n tool_matches.append(pattern)\n\n if tool_matches:\n predicted.add(tool)\n matched_patterns.append(f\"{tool}: {', '.join(tool_matches)}\")\n\n # Calculate confidence based on pattern matches\n total_words = len(prompt_lower.split())\n pattern_match_count = len(matched_patterns)\n\n # Base confidence on match density\n if total_words == 0:\n confidence = 0.5\n elif pattern_match_count == 0:\n # Only base tools predicted\n confidence = 0.5\n else:\n # More matches relative to prompt length = higher confidence\n confidence = min(0.9, 0.5 + (pattern_match_count / max(total_words / 5, 1)) * 0.4)\n\n # Adjust confidence based on prompt clarity\n if total_words \u003c 5:\n confidence *= 0.8 # Short prompts are less clear\n elif total_words > 20:\n confidence *= 0.95 # Very detailed prompts slightly less confident\n\n # Check capabilities if provided in envelope\n if envelope and \"capabilities\" in envelope:\n capabilities = envelope[\"capabilities\"]\n for skill_name, skill_info in capabilities.items():\n description = skill_info.get(\"description\", \"\").lower()\n\n # If skill description matches prompt, boost confidence\n if any(word in description for word in prompt_lower.split()):\n confidence = min(1.0, confidence + 0.1)\n\n # Data quality skills typically need more tools\n if \"quality\" in skill_name or \"governance\" in skill_name:\n predicted.update([\"glob\", \"grep\", \"write\"])\n matched_patterns.append(f\"skill_match: {skill_name}\")\n\n # ML skills need bash for model operations\n if \"ml\" in skill_name or \"machine\" in skill_name or \"forecast\" in skill_name:\n predicted.add(\"bash\")\n matched_patterns.append(f\"skill_match: {skill_name}\")\n\n # Generate reasoning\n if matched_patterns:\n reasoning = f\"Matched {len(matched_patterns)} patterns: {'; '.join(matched_patterns[:3])}\"\n if len(matched_patterns) > 3:\n reasoning += f\" and {len(matched_patterns) - 3} more\"\n else:\n reasoning = \"Using base Snowflake tools only - no specific patterns matched\"\n\n return {\n \"tools\": sorted(list(predicted)),\n \"confidence\": round(confidence, 2),\n \"reasoning\": reasoning\n }\n\n\ndef main():\n \"\"\"Main tool prediction function.\"\"\"\n parser = argparse.ArgumentParser(description=\"Predict required Cortex tools\")\n parser.add_argument(\"--prompt\", required=True, help=\"User prompt to analyze\")\n args = parser.parse_args()\n\n # Load capabilities\n capabilities = load_capabilities()\n envelope = {\"capabilities\": capabilities} if capabilities else None\n\n # Predict tools with confidence\n result = predict_tools(args.prompt, envelope)\n\n # Output as JSON\n print(json.dumps(result, indent=2))\n\n # Summary to stderr\n print(f\"\\nPredicted {len(result['tools'])} tools with {result['confidence']:.0%} confidence:\", file=sys.stderr)\n print(f\" Tools: {', '.join(result['tools'])}\", file=sys.stderr)\n print(f\" Reasoning: {result['reasoning']}\", file=sys.stderr)\n\n return 0\n\n\nif __name__ == \"__main__\":\n sys.exit(main())\n","content_type":"text/x-python; charset=utf-8","language":"python","size":5635,"content_sha256":"a6a552051b2ef299b041554c1117d824b932ec859b3d7711325bd42f169b6a42"},{"filename":"scripts/read_cortex_sessions.py","content":"#!/usr/bin/env python3\n\"\"\"\nReads recent Cortex Code session files for context enrichment.\n\"\"\"\n\nimport json\nimport sys\nimport argparse\nfrom pathlib import Path\nfrom datetime import datetime\n\nMAX_SESSION_BYTES = 5 * 1024 * 1024\n\n# Add parent directory to path for imports\nsys.path.insert(0, str(Path(__file__).parent.parent))\nfrom security.prompt_sanitizer import PromptSanitizer\n\n\ndef find_recent_sessions(limit=3):\n \"\"\"Find the most recent Cortex session files.\"\"\"\n sessions_dir = Path.home() / \".local/share/cortex/sessions\"\n\n if not sessions_dir.exists():\n print(f\"Sessions directory not found: {sessions_dir}\", file=sys.stderr)\n return []\n\n # Find all .jsonl session files\n session_files = sorted(\n [f for f in sessions_dir.glob(\"**/*.jsonl\")],\n key=lambda f: f.stat().st_mtime,\n reverse=True\n )\n\n return session_files[:limit]\n\n\ndef parse_session_file(session_path, sanitize=True):\n \"\"\"Parse a session JSONL file and extract key information.\n\n Args:\n session_path: Path to the session JSONL file\n sanitize: Whether to sanitize PII from text content (default: True)\n\n Returns:\n Dictionary with session data, or None on error\n \"\"\"\n try:\n if session_path.stat().st_size > MAX_SESSION_BYTES:\n print(f\"Skipping oversized session file: {session_path}\", file=sys.stderr)\n return None\n\n # Initialize sanitizer if needed\n sanitizer = PromptSanitizer() if sanitize else None\n\n session_data = {\n \"session_id\": None,\n \"timestamp\": session_path.stat().st_mtime,\n \"user_prompts\": [],\n \"assistant_responses\": [],\n \"tools_used\": [],\n \"result\": None\n }\n\n with open(session_path, 'r') as f:\n for line in f:\n if not line.strip():\n continue\n\n try:\n event = json.loads(line)\n event_type = event.get(\"type\")\n\n if event_type == \"system\" and event.get(\"subtype\") == \"init\":\n session_data[\"session_id\"] = event.get(\"session_id\")\n\n elif event_type == \"user\":\n # Check if this is a tool result or user message\n message = event.get(\"message\", {})\n content = message.get(\"content\", [])\n\n # Extract user text if present\n for item in content:\n if item.get(\"type\") == \"text\":\n text = item.get(\"text\", \"\")\n # Sanitize user prompts if enabled\n if sanitizer:\n text = sanitizer.sanitize(text)\n session_data[\"user_prompts\"].append(text)\n\n elif event_type == \"assistant\":\n message = event.get(\"message\", {})\n content = message.get(\"content\", [])\n\n for item in content:\n if item.get(\"type\") == \"text\":\n text = item.get(\"text\", \"\")\n # Sanitize assistant responses if enabled\n if sanitizer:\n text = sanitizer.sanitize(text)\n session_data[\"assistant_responses\"].append(text)\n elif item.get(\"type\") == \"tool_use\":\n tool_name = item.get(\"name\")\n if tool_name:\n session_data[\"tools_used\"].append(tool_name)\n\n elif event_type == \"result\":\n session_data[\"result\"] = event.get(\"result\")\n\n except json.JSONDecodeError:\n continue\n\n return session_data\n\n except Exception as e:\n print(f\"Error parsing session {session_path}: {e}\", file=sys.stderr)\n return None\n\n\ndef summarize_sessions(session_files, sanitize=True):\n \"\"\"Summarize recent Cortex sessions.\n\n Args:\n session_files: List of session file paths\n sanitize: Whether to sanitize PII from text content (default: True)\n\n Returns:\n List of session summary dictionaries\n \"\"\"\n summaries = []\n\n for session_path in session_files:\n session_data = parse_session_file(session_path, sanitize=sanitize)\n\n if not session_data:\n continue\n\n # Create a concise summary\n # Note: session_data already has sanitized content if sanitize=True\n summary = {\n \"file\": session_path.name,\n \"session_id\": session_data[\"session_id\"],\n \"time\": datetime.fromtimestamp(session_data[\"timestamp\"]).strftime(\"%Y-%m-%d %H:%M:%S\"),\n \"prompts_count\": len(session_data[\"user_prompts\"]),\n \"tools_used\": list(set(session_data[\"tools_used\"])),\n \"last_prompt\": session_data[\"user_prompts\"][-1] if session_data[\"user_prompts\"] else None,\n \"result_type\": type(session_data[\"result\"]).__name__ if session_data[\"result\"] else None\n }\n\n summaries.append(summary)\n\n return summaries\n\n\ndef main():\n \"\"\"Main function to read and summarize recent Cortex sessions.\"\"\"\n parser = argparse.ArgumentParser(description=\"Read recent Cortex sessions\")\n parser.add_argument(\"--limit\", type=int, default=3, help=\"Number of recent sessions to read\")\n parser.add_argument(\"--verbose\", action=\"store_true\", help=\"Include full session details\")\n parser.add_argument(\"--no-sanitize\", action=\"store_true\", help=\"Disable PII sanitization (for debugging)\")\n args = parser.parse_args()\n\n # Determine if sanitization should be enabled (default: True)\n sanitize = not args.no_sanitize\n\n # Find recent sessions\n session_files = find_recent_sessions(args.limit)\n\n if not session_files:\n print(\"No recent Cortex sessions found\", file=sys.stderr)\n return 0\n\n print(f\"Found {len(session_files)} recent sessions\", file=sys.stderr)\n\n # Summarize sessions with sanitization flag\n summaries = summarize_sessions(session_files, sanitize=sanitize)\n\n # Output JSON\n print(json.dumps(summaries, indent=2))\n\n return 0\n\n\nif __name__ == \"__main__\":\n sys.exit(main())\n","content_type":"text/x-python; charset=utf-8","language":"python","size":6417,"content_sha256":"6bf6e2e72cce6a0a2548119d512013309f57e837888075d1d401b659c4a3b0a5"},{"filename":"scripts/route_request.py","content":"#!/usr/bin/env python3\n\"\"\"\nLLM-based routing logic to determine if request should go to Cortex Code or Codex.\nUses semantic understanding rather than simple keyword matching.\n\"\"\"\n\nimport json\nimport sys\nimport argparse\nimport fnmatch\nimport re\nfrom pathlib import Path\nfrom typing import Optional, Dict, Any\n\n# Add parent directory to path for security imports\nsys.path.insert(0, str(Path(__file__).parent.parent))\nfrom security.config_manager import ConfigManager\nfrom security.cache_manager import CacheManager\n\n\n# Snowflake/Cortex indicators\nSNOWFLAKE_INDICATORS = [\n \"snowflake\", \"cortex\", \"warehouse\", \"snowpark\", \"data warehouse\",\n \"cortex ai\", \"cortex search\", \"cortex analyst\", \"dynamic table\",\n \"snowflake database\", \"snowflake schema\", \"snowflake table\",\n \"data governance\", \"data quality\", \"trust my data\",\n \"ml function\", \"classification\", \"forecasting\"\n]\n\n# Non-Snowflake indicators (route to Codex)\nSNOWFLAKE_CONTEXT_TERMS = [\"snowflake\", \"warehouse\", \"cortex\", \"schema\", \"table\", \"database\"]\nAMBIGUOUS_SNOWFLAKE_TERMS = [\"stream\", \"task\", \"stage\", \"pipe\"]\nPATH_TOKEN_PATTERN = re.compile(r'(?\u003c![\\w.-])(?:~/?|/|\\./|\\.\\./|[A-Za-z0-9_.-]+/)[A-Za-z0-9_./$~:-]+|(?\u003c![\\w.-])(?:\\.ssh|\\.aws|\\.snowflake|\\.env(?:\\.[\\w-]+)?|credentials\\.(?:json|ya?ml)|[A-Za-z0-9_.-]+_key\\.(?:p8|pem))(?![\\w.-])', re.IGNORECASE)\n\nCLAUDE_CODE_INDICATORS = [\n \"local file\", \"git\", \"github\", \"commit\", \"push\", \"pull request\",\n \"python script\", \"javascript\", \"react\", \"frontend\", \"backend\",\n \"postgres\", \"mysql\", \"mongodb\", \"redis\",\n \"docker\", \"kubernetes\", \"infrastructure\",\n \"read file\", \"write file\", \"edit file\", \"create file\"\n]\n\n# Backwards-compatible name used by shared tests and copied integrations.\nCODING_AGENT_INDICATORS = CLAUDE_CODE_INDICATORS\n\n\ndef load_cortex_capabilities():\n \"\"\"Load cached Cortex capabilities using CacheManager.\"\"\"\n try:\n # Get cache directory from config\n config_manager = ConfigManager()\n cache_dir_str = config_manager.get(\"security.cache_dir\")\n cache_dir = Path(cache_dir_str).expanduser()\n\n # Use CacheManager to read cache with integrity validation\n cache_manager = CacheManager(cache_dir)\n capabilities = cache_manager.read(\"cortex-capabilities\")\n\n if capabilities is None:\n print(\"Warning: Cortex capabilities not cached. Run discover_cortex.py first.\", file=sys.stderr)\n return {}\n\n return capabilities\n\n except Exception as e:\n print(f\"Warning: Failed to load Cortex capabilities from cache: {e}\", file=sys.stderr)\n print(\"Run discover_cortex.py to cache capabilities.\", file=sys.stderr)\n return {}\n\n\ndef analyze_with_llm_logic(prompt, capabilities):\n \"\"\"\n Analyze prompt using LLM-inspired logic.\n This is a deterministic approximation of what an LLM would consider.\n \"\"\"\n prompt_lower = prompt.lower()\n\n # Score based on indicators\n snowflake_score = 0\n claude_score = 0\n\n # Check for explicit Snowflake/Cortex mentions\n for indicator in SNOWFLAKE_INDICATORS:\n if indicator in prompt_lower:\n snowflake_score += 3 if indicator in [\"snowflake\", \"cortex\"] else 1\n\n # Ambiguous Snowflake object names only count with Snowflake context.\n if any(context in prompt_lower for context in SNOWFLAKE_CONTEXT_TERMS):\n for term in AMBIGUOUS_SNOWFLAKE_TERMS:\n if term in prompt_lower:\n snowflake_score += 1\n\n # Check for non-Snowflake indicators\n for indicator in CLAUDE_CODE_INDICATORS:\n if indicator in prompt_lower:\n claude_score += 2\n\n # Check against Cortex skill triggers\n for skill_name, skill_info in capabilities.items():\n for trigger in skill_info.get(\"triggers\", []):\n trigger_lower = trigger.lower()\n if trigger_lower in prompt_lower or any(word in prompt_lower for word in trigger_lower.split()):\n snowflake_score += 2\n break\n\n # SQL query detection\n sql_keywords = [\"select\", \"insert\", \"update\", \"delete\", \"create table\", \"alter\", \"drop\"]\n if any(kw in prompt_lower for kw in sql_keywords):\n # Could be any database, but check for Snowflake context\n if any(ind in prompt_lower for ind in [\"snowflake\", \"warehouse\", \"cortex\"]):\n snowflake_score += 3\n else:\n # Generic SQL, likely not Snowflake\n claude_score += 1\n\n # Data-related terms (ambiguous, need context)\n data_terms = [\"data quality\", \"schema\", \"table\", \"database\", \"query\"]\n data_term_count = sum(1 for term in data_terms if term in prompt_lower)\n if data_term_count >= 2:\n # Multiple data terms suggest database work\n # Check if Snowflake context exists\n if snowflake_score > 0:\n snowflake_score += 2\n\n # Calculate confidence\n total_score = snowflake_score + claude_score\n if total_score == 0:\n # No strong indicators, default to the host coding agent for safety.\n # Install scripts replace this placeholder with claude/codex/cursor.\n return \"__CODING_AGENT__\", 0.5\n\n confidence = max(snowflake_score, claude_score) / total_score\n\n if snowflake_score > claude_score:\n return \"cortex\", confidence\n else:\n return \"__CODING_AGENT__\", confidence\n\n\ndef check_credential_allowlist(\n prompt: str,\n config_path: Optional[Path] = None,\n org_policy_path: Optional[Path] = None\n) -> Dict[str, Any]:\n \"\"\"\n Check if prompt contains credential file paths from the allowlist.\n\n This function runs before routing analysis to block prompts that reference\n credential files, regardless of whether they would be routed to Cortex or Codex.\n\n Args:\n prompt: User prompt to check\n config_path: Path to user config file (optional)\n org_policy_path: Path to organization policy file (optional)\n\n Returns:\n Dict with blocking decision:\n - blocked: True if credential detected, False otherwise\n - route: \"blocked\" if blocked, None otherwise\n - confidence: 1.0 if blocked (100% confident in blocking)\n - reason: Human-readable reason for blocking\n - pattern_matched: The allowlist pattern that matched\n \"\"\"\n # Initialize ConfigManager with optional config paths\n config_manager = ConfigManager(\n config_path=config_path,\n org_policy_path=org_policy_path\n )\n\n # Load credential allowlist\n credential_allowlist = config_manager.get(\"security.credential_file_allowlist\")\n\n prompt_tokens = PATH_TOKEN_PATTERN.findall(prompt)\n normalized_tokens = []\n for token in prompt_tokens:\n normalized_tokens.append(token)\n if token.startswith(\"~\"):\n normalized_tokens.append(token.replace(\"~\", str(Path.home()), 1))\n\n for pattern in credential_allowlist:\n expanded_pattern = str(Path(pattern).expanduser())\n candidate_patterns = [pattern, expanded_pattern]\n if pattern.startswith(\"~/**/\"):\n candidate_patterns.append(\"**/\" + pattern.split(\"~/**/\", 1)[1])\n for token in normalized_tokens:\n token_lower = token.lower()\n for candidate_pattern in candidate_patterns:\n pattern_lower = candidate_pattern.lower()\n pattern_dir = pattern_lower.split(\"*\")[0].rstrip(\"/\")\n if (\n fnmatch.fnmatch(token_lower, pattern_lower)\n or fnmatch.fnmatch(f\"*/{token_lower}\", pattern_lower)\n or (token_lower in {\".ssh\", \".aws\", \".snowflake\"} and pattern_dir.endswith(token_lower))\n ):\n return {\n \"blocked\": True,\n \"route\": \"blocked\",\n \"confidence\": 1.0,\n \"reason\": f\"Prompt contains credential file path from allowlist\",\n \"pattern_matched\": pattern\n }\n\n # No credentials detected\n return {\n \"blocked\": False\n }\n\n\ndef main():\n \"\"\"Main routing function.\"\"\"\n parser = argparse.ArgumentParser(description=\"Route request to Cortex or Codex\")\n parser.add_argument(\"--prompt\", required=True, help=\"User prompt to analyze\")\n parser.add_argument(\"--config\", help=\"Path to user config file\")\n parser.add_argument(\"--org-policy\", help=\"Path to organization policy file\")\n args = parser.parse_args()\n\n # Step 1: Check credential allowlist BEFORE routing\n config_path = Path(args.config) if args.config else None\n org_policy_path = Path(args.org_policy) if args.org_policy else None\n\n credential_check = check_credential_allowlist(\n args.prompt,\n config_path,\n org_policy_path\n )\n\n # If blocked by credential check, return immediately\n if credential_check.get(\"blocked\"):\n print(json.dumps(credential_check, indent=2))\n print(f\"\\n⛔ BLOCKED: Credential file detected\", file=sys.stderr)\n print(f\" Pattern: {credential_check['pattern_matched']}\", file=sys.stderr)\n print(f\" Reason: {credential_check['reason']}\", file=sys.stderr)\n sys.exit(0)\n\n # Step 2: Load Cortex capabilities\n capabilities = load_cortex_capabilities()\n\n # Step 3: Analyze prompt for routing\n route, confidence = analyze_with_llm_logic(args.prompt, capabilities)\n\n # Step 4: Output decision\n result = {\n \"route\": route,\n \"confidence\": confidence,\n \"reasoning\": f\"Routed to {route} with {confidence:.2%} confidence\"\n }\n\n print(json.dumps(result, indent=2))\n\n print(f\"\\n→ Route to: {route.upper()}\", file=sys.stderr)\n print(f\" Confidence: {confidence:.2%}\", file=sys.stderr)\n\n sys.exit(0)\n\n\nif __name__ == \"__main__\":\n main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":9751,"content_sha256":"e8bd4fd2a4498eae018e8ecaeb9259b3d0b01dada520013d023ca894d3d4736e"},{"filename":"scripts/security_wrapper.py","content":"#!/usr/bin/env python3\n\"\"\"\nSecurity wrapper orchestrator for cortex-code skill.\n\nCoordinates all security components:\n- ConfigManager: Load and validate configuration\n- AuditLogger: Log all executions\n- CacheManager: Secure caching\n- PromptSanitizer: Remove PII and detect injection\n- ApprovalHandler: Tool prediction and user approval\n\nThis is the main entry point for secure Cortex execution.\n\"\"\"\n\nimport argparse\nimport fnmatch\nimport json\nimport re\nimport sys\nimport os\nfrom pathlib import Path\nfrom typing import Optional, Dict, Any\n\n# Add parent directories to path\nsys.path.insert(0, str(Path(__file__).parent.parent))\n\nfrom security.config_manager import ConfigManager\nfrom security.audit_logger import AuditLogger\nfrom security.cache_manager import CacheManager\nfrom security.prompt_sanitizer import PromptSanitizer\nfrom security.approval_handler import ApprovalHandler\n\n# Import routing functions\nsys.path.insert(0, str(Path(__file__).parent))\nfrom route_request import analyze_with_llm_logic, load_cortex_capabilities\nfrom execute_cortex import execute_cortex_streaming\n\n\ndef _log_audit_event(audit_logger, **kwargs):\n \"\"\"Best-effort audit logging helper.\"\"\"\n try:\n return audit_logger.log_execution(**kwargs), None\n except Exception as exc:\n print(f\"Warning: failed to write audit log: {exc}\", file=sys.stderr)\n return None, str(exc)\n\n\nPATH_TOKEN_PATTERN = re.compile(r'(?\u003c![\\w.-])(?:~/?|/|\\./|\\.\\./|[A-Za-z0-9_.-]+/)[A-Za-z0-9_./$~:-]+|(?\u003c![\\w.-])(?:\\.ssh|\\.aws|\\.snowflake|\\.env(?:\\.[\\w-]+)?|credentials\\.(?:json|ya?ml)|[A-Za-z0-9_.-]+_key\\.(?:p8|pem))(?![\\w.-])', re.IGNORECASE)\n\n\ndef execute_with_security(\n prompt: str,\n config_path: Optional[str] = None,\n org_policy_path: Optional[str] = None,\n dry_run: bool = False,\n envelope: Optional[Dict[str, Any]] = None,\n mock_user_approval: Optional[str] = None\n) -> Dict[str, Any]:\n \"\"\"\n Execute prompt with full security orchestration.\n\n This function:\n 1. Loads configuration (with org policy override)\n 2. Initializes all security components\n 3. Sanitizes prompt if enabled\n 4. Determines approval mode\n 5. In dry-run mode: returns initialization status\n 6. In live mode: Full execution with approval flow\n\n Args:\n prompt: User prompt to execute\n config_path: Path to user config file (optional)\n org_policy_path: Path to organization policy file (optional)\n dry_run: If True, only initialize and validate (don't execute)\n envelope: Cortex envelope dict (optional)\n mock_user_approval: For testing - \"approve\" or \"deny\" (optional)\n\n Returns:\n Dict with execution results or initialization status\n \"\"\"\n # Step 1: Load configuration\n config_path_obj = Path(config_path) if config_path else None\n org_policy_path_obj = Path(org_policy_path) if org_policy_path else None\n\n config_manager = ConfigManager(\n config_path=config_path_obj,\n org_policy_path=org_policy_path_obj\n )\n\n # Extract config values\n approval_mode = config_manager.get(\"security.approval_mode\")\n audit_log_path = Path(config_manager.get(\"security.audit_log_path\"))\n audit_log_rotation = config_manager.get(\"security.audit_log_rotation\")\n audit_log_retention = config_manager.get(\"security.audit_log_retention\")\n cache_dir = Path(config_manager.get(\"security.cache_dir\"))\n sanitize_enabled = config_manager.get(\"security.sanitize_conversation_history\")\n confidence_threshold = config_manager.get(\"security.tool_prediction_confidence_threshold\")\n allowed_envelopes = config_manager.get(\"security.allowed_envelopes\")\n\n # Step 2: Initialize security components\n audit_logger = AuditLogger(\n log_path=audit_log_path,\n rotation_size=audit_log_rotation,\n retention_days=audit_log_retention\n )\n\n cache_manager = CacheManager(cache_dir=cache_dir)\n\n prompt_sanitizer = PromptSanitizer()\n\n approval_handler = ApprovalHandler(confidence_threshold=confidence_threshold)\n\n # Step 3: Sanitize prompt if enabled\n sanitized_prompt = prompt\n if sanitize_enabled:\n sanitized_prompt = prompt_sanitizer.sanitize(prompt)\n if sanitized_prompt == \"[POTENTIAL INJECTION DETECTED - REMOVED]\":\n return {\n \"status\": \"blocked\",\n \"reason\": \"Prompt injection attempt detected\",\n \"message\": \"Cannot route prompts containing prompt injection attempts\",\n \"sanitized_prompt\": sanitized_prompt\n }\n\n envelope_mode = \"RW\"\n if isinstance(envelope, dict):\n envelope_mode = envelope.get(\"mode\") or envelope.get(\"type\") or \"RW\"\n elif isinstance(envelope, str):\n envelope_mode = envelope\n\n if envelope_mode not in allowed_envelopes:\n return {\n \"status\": \"blocked\",\n \"reason\": f\"Envelope {envelope_mode} is not allowed by configuration\",\n \"allowed_envelopes\": allowed_envelopes,\n \"requested_envelope\": envelope_mode,\n }\n\n # Step 4: Check credential file allowlist (on original prompt)\n credential_allowlist = config_manager.get(\"security.credential_file_allowlist\")\n prompt_tokens = PATH_TOKEN_PATTERN.findall(prompt)\n normalized_tokens = []\n for token in prompt_tokens:\n normalized_tokens.append(token)\n if token.startswith(\"~\"):\n normalized_tokens.append(token.replace(\"~\", str(Path.home()), 1))\n for pattern in credential_allowlist:\n expanded_pattern = str(Path(pattern).expanduser())\n candidate_patterns = [pattern, expanded_pattern]\n if pattern.startswith(\"~/**/\"):\n candidate_patterns.append(\"**/\" + pattern.split(\"~/**/\", 1)[1])\n for token in normalized_tokens:\n token_lower = token.lower()\n for candidate_pattern in candidate_patterns:\n pattern_lower = candidate_pattern.lower()\n pattern_dir = pattern_lower.split(\"*\")[0].rstrip(\"/\")\n if (\n fnmatch.fnmatch(token_lower, pattern_lower)\n or fnmatch.fnmatch(f\"*/{token_lower}\", pattern_lower)\n or (token_lower in {\".ssh\", \".aws\", \".snowflake\"} and pattern_dir.endswith(token_lower))\n ):\n return {\n \"status\": \"blocked\",\n \"reason\": \"Prompt contains credential file path from allowlist\",\n \"pattern_matched\": pattern,\n \"message\": \"Cannot route prompts containing credential file paths for security\"\n }\n\n # Step 5: Determine routing (cortex vs claude) on sanitized prompt\n capabilities = load_cortex_capabilities()\n route_decision, route_confidence = analyze_with_llm_logic(sanitized_prompt, capabilities)\n\n # Step 6: Determine approval mode\n # In prompt mode, user must approve tools\n # In auto mode, tools are auto-approved\n # In deny mode, execution is blocked\n\n # Step 7: Dry-run mode - return initialization status\n if dry_run:\n return {\n \"status\": \"initialized\",\n \"dry_run\": True,\n \"sanitized_prompt\": sanitized_prompt,\n \"routing\": {\n \"decision\": route_decision,\n \"confidence\": route_confidence\n },\n \"config\": {\n \"approval_mode\": approval_mode,\n \"audit_log_path\": str(audit_log_path),\n \"cache_dir\": str(cache_dir),\n \"sanitize_enabled\": sanitize_enabled,\n \"confidence_threshold\": confidence_threshold,\n \"allowed_envelopes\": allowed_envelopes\n },\n \"audit_logger\": str(type(audit_logger).__name__),\n \"cache_manager\": str(type(cache_manager).__name__),\n \"prompt_sanitizer\": str(type(prompt_sanitizer).__name__),\n \"approval_handler\": str(type(approval_handler).__name__)\n }\n\n # Step 8: Full execution flow\n # Route to Coding Agent for non-Snowflake requests\n if route_decision == \"__CODING_AGENT__\":\n return {\n \"status\": \"routed_to_coding_agent\",\n \"message\": \"Request routed to coding agent for local handling\",\n \"routing\": {\"decision\": route_decision, \"confidence\": route_confidence}\n }\n\n # Step 9: Tool prediction for Cortex execution\n prediction = approval_handler.predict_tools(sanitized_prompt, envelope)\n predicted_tools = prediction[\"tools\"]\n tool_confidence = prediction[\"confidence\"]\n\n # Step 10: Handle approval mode\n allowed_tools = []\n approval_result = None\n\n if approval_mode == \"prompt\":\n # Prompt mode: require user approval\n if mock_user_approval:\n # Testing mode - mock approval\n if mock_user_approval == \"approve\":\n allowed_tools = predicted_tools\n elif mock_user_approval == \"deny\":\n return {\n \"status\": \"denied\",\n \"message\": \"User denied execution\",\n \"predicted_tools\": predicted_tools\n }\n else:\n # Real mode - format approval prompt\n approval_prompt = approval_handler.format_approval_prompt(\n predicted_tools,\n tool_confidence,\n envelope,\n prediction.get(\"reasoning\", \"\")\n )\n\n approval_result = {\n \"status\": \"awaiting_approval\",\n \"approval_prompt\": approval_prompt,\n \"predicted_tools\": predicted_tools,\n \"confidence\": tool_confidence,\n \"envelope\": envelope\n }\n audit_id, audit_error = _log_audit_event(\n audit_logger,\n event_type=\"cortex_approval_requested\",\n user=os.environ.get(\"USER\", \"unknown\"),\n routing={\"decision\": route_decision, \"confidence\": route_confidence},\n execution={\n \"envelope\": envelope,\n \"approval_mode\": approval_mode,\n \"auto_approved\": False,\n \"predicted_tools\": predicted_tools,\n \"allowed_tools\": []\n },\n result={\"status\": \"awaiting_approval\"},\n security={\n \"sanitized\": sanitize_enabled,\n \"pii_removed\": sanitize_enabled and prompt != sanitized_prompt\n }\n )\n approval_result[\"audit_id\"] = audit_id\n approval_result[\"audit_error\"] = audit_error\n return approval_result\n\n elif approval_mode == \"auto\":\n # Auto mode: auto-approve all tools\n allowed_tools = predicted_tools\n\n elif approval_mode == \"envelope_only\":\n # Envelope only mode: no tool prediction\n allowed_tools = None # None means rely on envelope only\n\n # Step 11: Execute with Cortex using the sanitized prompt.\n if mock_user_approval:\n execution_result = {\n \"status\": \"success\",\n \"message\": \"Execution simulated for mocked approval\",\n \"tools_used\": allowed_tools or [\"envelope-controlled\"],\n }\n else:\n execution_result = execute_cortex_streaming(\n prompt=sanitized_prompt,\n envelope=envelope_mode,\n approval_mode=approval_mode,\n allowed_tools=allowed_tools,\n timeout_seconds=int(config_manager.get(\"security.execution_timeout_seconds\", 5)),\n deploy_confirmed=bool(config_manager.get(\"security.deploy_envelope_confirmation\", True) and envelope_mode == \"DEPLOY\"),\n )\n execution_result.setdefault(\"status\", \"success\" if not execution_result.get(\"error\") else \"error\")\n execution_result.setdefault(\"tools_used\", allowed_tools or [\"envelope-controlled\"])\n\n # Step 12: Audit logging\n audit_id, audit_error = _log_audit_event(\n audit_logger,\n event_type=\"cortex_execution\",\n user=os.environ.get(\"USER\", \"unknown\"),\n routing={\"decision\": route_decision, \"confidence\": route_confidence},\n execution={\n \"envelope\": envelope,\n \"approval_mode\": approval_mode,\n \"auto_approved\": approval_mode in [\"auto\", \"envelope_only\"],\n \"predicted_tools\": predicted_tools,\n \"allowed_tools\": allowed_tools\n },\n result=execution_result,\n security={\n \"sanitized\": sanitize_enabled,\n \"pii_removed\": sanitize_enabled and prompt != sanitized_prompt\n }\n )\n\n # Step 13: Cache result (optional - for future optimization)\n # For now, skip caching\n\n return {\n \"status\": \"executed\",\n \"audit_id\": audit_id,\n \"audit_error\": audit_error,\n \"routing\": {\"decision\": route_decision, \"confidence\": route_confidence},\n \"approval_mode\": approval_mode,\n \"predicted_tools\": predicted_tools,\n \"allowed_tools\": allowed_tools,\n \"result\": execution_result,\n \"security\": {\n \"sanitized\": sanitize_enabled,\n \"pii_removed\": sanitize_enabled and prompt != sanitized_prompt\n }\n }\n\n\ndef main():\n \"\"\"CLI entry point for security wrapper.\"\"\"\n parser = argparse.ArgumentParser(\n description=\"Security wrapper for cortex-code skill\"\n )\n parser.add_argument(\n \"--prompt\",\n required=True,\n help=\"User prompt to execute\"\n )\n parser.add_argument(\n \"--config\",\n help=\"Path to user config file\"\n )\n parser.add_argument(\n \"--org-policy\",\n help=\"Path to organization policy file\"\n )\n parser.add_argument(\n \"--dry-run\",\n action=\"store_true\",\n help=\"Dry-run mode: initialize and validate only\"\n )\n parser.add_argument(\n \"--envelope\",\n help=\"Cortex envelope JSON string\"\n )\n\n args = parser.parse_args()\n\n # Parse envelope if provided\n envelope = None\n if args.envelope:\n try:\n envelope = json.loads(args.envelope)\n except json.JSONDecodeError as e:\n print(json.dumps({\n \"status\": \"error\",\n \"message\": f\"Invalid envelope JSON: {e}\"\n }))\n sys.exit(1)\n\n # Execute with security\n try:\n result = execute_with_security(\n prompt=args.prompt,\n config_path=args.config,\n org_policy_path=args.org_policy,\n dry_run=args.dry_run,\n envelope=envelope\n )\n print(json.dumps(result, indent=2))\n except Exception as e:\n print(json.dumps({\n \"status\": \"error\",\n \"message\": str(e)\n }))\n sys.exit(1)\n\n\nif __name__ == \"__main__\":\n main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":14780,"content_sha256":"54bf562b828aafd8c5ec18165908a5cbc5934d5f65f49a218efd6d245a99ca69"},{"filename":"security/__init__.py","content":"\"\"\"Security layer for cortex-code skill.\"\"\"\n\n__version__ = \"1.0.0\"\n","content_type":"text/x-python; charset=utf-8","language":"python","size":67,"content_sha256":"511a10585394004e1ee9b7209034b15a2f0371501ee063dc174506f74ec3573e"},{"filename":"security/approval_handler.py","content":"#!/usr/bin/env python3\n\"\"\"\nApproval handler for tool prediction and user approval flow.\nPredicts which tools Cortex needs and formats approval prompts for users.\n\"\"\"\n\nfrom dataclasses import dataclass\nfrom typing import List, Dict, Any, Optional\nimport sys\nfrom pathlib import Path\n\n# Add scripts directory to path for predict_tools import\nscripts_dir = Path(__file__).parent.parent / \"scripts\"\nsys.path.insert(0, str(scripts_dir))\n\nfrom predict_tools import predict_tools as predict_tools_func\n\n\n@dataclass\nclass ApprovalResult:\n \"\"\"Result of approval process.\"\"\"\n approved: bool\n allowed_tools: List[str]\n user_response: str\n\n\nclass ApprovalHandler:\n \"\"\"\n Handles tool prediction and user approval flow.\n\n Predicts which tools Cortex needs based on user prompts,\n formats approval prompts with confidence scores and warnings,\n and parses user responses.\n \"\"\"\n\n def __init__(self, confidence_threshold: float = 0.7):\n \"\"\"\n Initialize approval handler.\n\n Args:\n confidence_threshold: Minimum confidence for predictions (default 0.7)\n \"\"\"\n self.confidence_threshold = confidence_threshold\n\n def predict_tools(self, prompt: str, envelope: Dict[str, Any]) -> Dict[str, Any]:\n \"\"\"\n Predict which tools will be needed for the given prompt.\n\n Args:\n prompt: User prompt to analyze\n envelope: Request envelope with capabilities and context\n\n Returns:\n dict with:\n - tools: list of predicted tool names\n - confidence: float 0-1 indicating prediction confidence\n - reasoning: str explaining the prediction\n \"\"\"\n return predict_tools_func(prompt, envelope)\n\n def format_approval_prompt(\n self,\n tools: List[str],\n confidence: float,\n envelope: Dict[str, Any],\n reasoning: str\n ) -> str:\n \"\"\"\n Format approval prompt for user.\n\n Args:\n tools: List of predicted tool names\n confidence: Prediction confidence (0-1)\n envelope: Request envelope with user_prompt and context\n reasoning: Explanation of tool prediction\n\n Returns:\n Formatted approval prompt string\n \"\"\"\n user_prompt = envelope.get(\"user_prompt\", \"Unknown request\")\n\n # Build approval prompt\n lines = []\n lines.append(\"=\" * 70)\n lines.append(\"CORTEX TOOL APPROVAL REQUEST\")\n lines.append(\"=\" * 70)\n lines.append(\"\")\n lines.append(f\"User Request: {user_prompt}\")\n lines.append(\"\")\n lines.append(f\"Predicted Tools ({len(tools)}):\")\n for tool in tools:\n lines.append(f\" - {tool}\")\n lines.append(\"\")\n lines.append(f\"Prediction Confidence: {confidence:.0%}\")\n lines.append(f\"Reasoning: {reasoning}\")\n lines.append(\"\")\n\n # Add warning if confidence is below threshold\n if confidence \u003c self.confidence_threshold:\n lines.append(\"⚠️ WARNING: Low confidence prediction!\")\n lines.append(f\" Confidence {confidence:.0%} is below threshold {self.confidence_threshold:.0%}\")\n lines.append(\" Tool predictions may be uncertain or incomplete.\")\n lines.append(\"\")\n\n lines.append(\"=\" * 70)\n lines.append(\"APPROVAL OPTIONS:\")\n lines.append(\" approve - Allow these specific tools for this request\")\n lines.append(\" approve_all - Allow all tools (bypass future approvals)\")\n lines.append(\" deny - Reject this request\")\n lines.append(\"=\" * 70)\n lines.append(\"\")\n lines.append(\"Your response: \")\n\n return \"\\n\".join(lines)\n\n def parse_user_response(self, response: str) -> ApprovalResult:\n \"\"\"\n Parse user response to approval prompt.\n\n Args:\n response: User's response string\n\n Returns:\n ApprovalResult with approval decision and allowed tools\n \"\"\"\n response_lower = response.strip().lower()\n\n if response_lower == \"approve\":\n return ApprovalResult(\n approved=True,\n allowed_tools=[], # Will be filled by caller with predicted tools\n user_response=\"approve\"\n )\n elif response_lower == \"approve_all\":\n return ApprovalResult(\n approved=True,\n allowed_tools=[\"*\"], # Wildcard for all tools\n user_response=\"approve_all\"\n )\n elif response_lower == \"deny\":\n return ApprovalResult(\n approved=False,\n allowed_tools=[],\n user_response=\"deny\"\n )\n else:\n # Unknown response - treat as deny for safety\n return ApprovalResult(\n approved=False,\n allowed_tools=[],\n user_response=response\n )\n","content_type":"text/x-python; charset=utf-8","language":"python","size":4973,"content_sha256":"77228d5de59e220b5052866334e3b503c7d2c1eeb07e8b49fddc13f609611c29"},{"filename":"security/audit_logger.py","content":"\"\"\"Structured JSON audit logging with rotation.\"\"\"\nimport hashlib\nimport json\nimport os\nimport uuid\nfrom datetime import datetime, timezone\nfrom pathlib import Path\nfrom typing import Any, Dict, Optional\n\n\nclass AuditLogger:\n \"\"\"Audit logger with structured JSON format and file rotation.\n\n Note: This implementation is designed for single-process use only.\n Concurrent writes from multiple processes may result in interleaved\n JSON lines or race conditions during rotation. For multi-process\n scenarios, consider using a log aggregation service or file locking.\n \"\"\"\n\n VERSION = \"2.0.0\"\n\n def __init__(\n self,\n log_path: Path,\n rotation_size: str = \"10MB\",\n retention_days: int = 30\n ):\n \"\"\"Initialize audit logger.\n\n Args:\n log_path: Path to audit log file\n rotation_size: Size threshold for rotation (e.g., \"10MB\", \"1GB\")\n retention_days: Days to retain rotated logs (NOT YET IMPLEMENTED)\n \"\"\"\n self.log_path = Path(log_path)\n self.rotation_size = self._parse_size(rotation_size)\n self.retention_days = retention_days\n self.initialization_error: Optional[str] = None\n # TODO: Implement cleanup of rotated files older than retention_days\n\n try:\n self.log_path.parent.mkdir(parents=True, exist_ok=True)\n\n if not self.log_path.exists():\n self.log_path.touch(mode=0o600)\n else:\n os.chmod(self.log_path, 0o600)\n except OSError as exc:\n self.initialization_error = str(exc)\n\n def log_execution(\n self,\n event_type: str,\n user: str,\n routing: Dict[str, Any],\n execution: Dict[str, Any],\n result: Dict[str, Any],\n session_id: Optional[str] = None,\n cortex_session_id: Optional[str] = None,\n security: Optional[Dict[str, Any]] = None\n ) -> str:\n \"\"\"Log a cortex execution event.\"\"\"\n if self.initialization_error:\n raise OSError(self.initialization_error)\n\n audit_id = str(uuid.uuid4())\n\n entry = {\n \"timestamp\": datetime.now(timezone.utc).isoformat(),\n \"version\": self.VERSION,\n \"audit_id\": audit_id,\n \"event_type\": event_type,\n \"user\": user,\n \"session_id\": session_id,\n \"cortex_session_id\": cortex_session_id,\n \"routing\": routing,\n \"execution\": execution,\n \"result\": result,\n \"security\": security or {}\n }\n\n entry[\"prev_hash\"] = self._last_entry_hash()\n entry[\"entry_hash\"] = self._entry_hash(entry)\n\n self._write_entry(entry)\n self._rotate_if_needed()\n\n return audit_id\n\n def _entry_hash(self, entry: Dict[str, Any]) -> str:\n \"\"\"Hash a canonical audit entry for tamper-evident chaining.\"\"\"\n payload = json.dumps(entry, sort_keys=True, separators=(\",\", \":\"))\n return hashlib.sha256(payload.encode()).hexdigest()\n\n def _last_entry_hash(self) -> Optional[str]:\n \"\"\"Return the previous entry hash if the audit log has entries.\"\"\"\n if not self.log_path.exists():\n return None\n try:\n last_line = None\n with open(self.log_path, 'r') as f:\n for line in f:\n if line.strip():\n last_line = line\n if not last_line:\n return None\n return json.loads(last_line).get(\"entry_hash\")\n except (OSError, json.JSONDecodeError):\n return None\n\n def _write_entry(self, entry: Dict[str, Any]) -> None:\n \"\"\"Write entry to log file as JSON.\n\n Opens file for each write to avoid holding file handles open long-term.\n This trades some efficiency for simplicity and crash-safety (no buffering).\n If file was deleted externally, it will be recreated with default permissions.\n \"\"\"\n with open(self.log_path, 'a') as f:\n f.write(json.dumps(entry) + '\\n')\n\n def _parse_size(self, size_str: str) -> int:\n \"\"\"Parse size string like '10MB' to bytes.\"\"\"\n size_str = size_str.upper()\n multipliers = {\n 'KB': 1024,\n 'MB': 1024 * 1024,\n 'GB': 1024 * 1024 * 1024\n }\n\n for suffix, multiplier in multipliers.items():\n if size_str.endswith(suffix):\n try:\n value = float(size_str[:-len(suffix)])\n return int(value * multiplier)\n except ValueError:\n pass\n\n # Default to bytes\n try:\n return int(size_str)\n except ValueError:\n return 10 * 1024 * 1024 # Default 10MB\n\n def _rotate_if_needed(self) -> None:\n \"\"\"Rotate log file if exceeds size limit.\"\"\"\n if not self.log_path.exists():\n return\n\n size = self.log_path.stat().st_size\n if size >= self.rotation_size:\n # Rotate: rename current to .1, .1 to .2, etc.\n timestamp = datetime.now(timezone.utc).strftime(\"%Y%m%d_%H%M%S\")\n rotated_path = self.log_path.with_suffix(f\".{timestamp}.log\")\n self.log_path.rename(rotated_path)\n\n # Create new log file\n self.log_path.touch(mode=0o600)\n","content_type":"text/x-python; charset=utf-8","language":"python","size":5349,"content_sha256":"4f0fa83960944deb0601ae0599b56e19621c222640f31387de86a5a69c365480"},{"filename":"security/cache_manager.py","content":"\"\"\"Secure cache manager with integrity validation.\"\"\"\nimport hashlib\nimport hmac\nimport json\nimport os\nimport time\nimport warnings\nfrom datetime import datetime, timezone\nfrom pathlib import Path\nfrom typing import Any, Optional\n\n\nclass CacheManager:\n \"\"\"Secure cache manager with fingerprint validation.\"\"\"\n\n VERSION = \"2.0.0\"\n\n def __init__(self, cache_dir: Path):\n \"\"\"Initialize cache manager.\"\"\"\n self.cache_dir = Path(cache_dir)\n self.cache_dir.mkdir(parents=True, exist_ok=True)\n\n # Set directory permissions to 0700 (owner only). Some managed or\n # sandboxed filesystems deny chmod on existing home-cache directories;\n # keep the cache usable rather than failing security-wrapper startup.\n try:\n os.chmod(self.cache_dir, 0o700)\n except PermissionError as exc:\n warnings.warn(\n f\"Could not set secure permissions on cache directory {self.cache_dir}: {exc}\",\n RuntimeWarning,\n stacklevel=2,\n )\n\n def _signature_key(self) -> bytes:\n \"\"\"Return key material for cache tamper detection.\"\"\"\n return os.environ.get(\n \"CORTEX_CODE_CACHE_HMAC_KEY\",\n f\"cortex-cache:{self.cache_dir}\"\n ).encode()\n\n def _calculate_signature(self, cache_entry: dict) -> str:\n \"\"\"Calculate HMAC over stable cache fields.\"\"\"\n signed_payload = {\n \"version\": cache_entry.get(\"version\"),\n \"created_at\": cache_entry.get(\"created_at\"),\n \"expires_at\": cache_entry.get(\"expires_at\"),\n \"data\": cache_entry.get(\"data\"),\n \"fingerprint\": cache_entry.get(\"fingerprint\"),\n }\n payload = json.dumps(signed_payload, sort_keys=True, separators=(\",\", \":\"))\n return hmac.new(self._signature_key(), payload.encode(), hashlib.sha256).hexdigest()\n\n def _validate_key(self, key: str) -> None:\n \"\"\"Validate cache key to prevent path traversal.\"\"\"\n if not key:\n raise ValueError(\"Cache key cannot be empty\")\n\n # Allow only alphanumeric, underscore, hyphen, and dot\n import re\n if not re.match(r'^[a-zA-Z0-9_.-]+

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

, key):\n raise ValueError(\n f\"Invalid cache key: {key}. \"\n f\"Only alphanumeric characters, underscores, hyphens, and dots are allowed.\"\n )\n\n # Prevent path traversal\n if '..' in key or '/' in key or '\\\\' in key:\n raise ValueError(f\"Invalid cache key: {key}. Path traversal not allowed.\")\n\n def write(self, key: str, data: Any, ttl: int = 86400) -> None:\n \"\"\"Write data to cache with TTL and fingerprint.\"\"\"\n self._validate_key(key)\n\n cache_entry = {\n \"version\": self.VERSION,\n \"created_at\": datetime.now(timezone.utc).isoformat(),\n \"expires_at\": time.time() + ttl,\n \"data\": data\n }\n\n # Calculate fingerprint\n data_str = json.dumps(data, sort_keys=True)\n fingerprint = hashlib.sha256(data_str.encode()).hexdigest()\n cache_entry[\"fingerprint\"] = fingerprint\n cache_entry[\"signature\"] = self._calculate_signature(cache_entry)\n\n # Write to file\n cache_file = self.cache_dir / f\"{key}.json\"\n with open(cache_file, 'w') as f:\n json.dump(cache_entry, f, indent=2)\n\n # Set file permissions to 0600 (owner read/write only)\n os.chmod(cache_file, 0o600)\n\n def read(self, key: str) -> Optional[Any]:\n \"\"\"Read data from cache with validation.\"\"\"\n self._validate_key(key)\n\n cache_file = self.cache_dir / f\"{key}.json\"\n\n if not cache_file.exists():\n return None\n\n try:\n with open(cache_file, 'r') as f:\n cache_entry = json.load(f)\n\n # Check expiration\n if cache_entry[\"expires_at\"] \u003c= time.time():\n # Expired - delete and return None\n cache_file.unlink(missing_ok=True)\n return None\n\n # Validate fingerprint\n data = cache_entry[\"data\"]\n data_str = json.dumps(data, sort_keys=True)\n expected_fingerprint = hashlib.sha256(data_str.encode()).hexdigest()\n\n if cache_entry[\"fingerprint\"] != expected_fingerprint:\n # Tampered - delete and return None\n cache_file.unlink(missing_ok=True)\n return None\n\n expected_signature = self._calculate_signature(cache_entry)\n if cache_entry.get(\"signature\") != expected_signature:\n # Tampered - delete and return None\n cache_file.unlink(missing_ok=True)\n return None\n\n return data\n\n except (json.JSONDecodeError, KeyError, FileNotFoundError, OSError):\n # Corrupted cache - delete and return None\n cache_file.unlink(missing_ok=True)\n return None\n\n def clear(self, key: Optional[str] = None) -> None:\n \"\"\"Clear cache entry or all entries.\"\"\"\n if key:\n self._validate_key(key)\n cache_file = self.cache_dir / f\"{key}.json\"\n if cache_file.exists():\n cache_file.unlink(missing_ok=True)\n else:\n # Clear all cache files\n for cache_file in self.cache_dir.glob(\"*.json\"):\n cache_file.unlink(missing_ok=True)\n","content_type":"text/x-python; charset=utf-8","language":"python","size":5410,"content_sha256":"32e8638c3cf97d1aca84329c707bcff7a207a470e76634c0473501ed193d9bb0"},{"filename":"security/config_manager.py","content":"\"\"\"Configuration manager with 3-layer precedence.\"\"\"\nimport copy\nimport os\nimport sys\nfrom pathlib import Path\nfrom typing import Any, Optional, Dict\nimport yaml\n\nclass ConfigValidationError(Exception):\n \"\"\"Raised when configuration validation fails.\"\"\"\n pass\n\n\nclass ConfigManager:\n \"\"\"Manages security configuration with precedence: org policy > user config > defaults.\"\"\"\n\n DEFAULT_CONFIG = {\n \"security\": {\n \"approval_mode\": \"prompt\",\n \"tool_prediction_confidence_threshold\": 0.7,\n \"allow_tool_expansion\": True,\n \"audit_log_path\": \"~/.__CODING_AGENT__/skills/cortex-code/audit.log\",\n \"audit_log_rotation\": \"10MB\",\n \"audit_log_retention\": 30,\n \"sanitize_conversation_history\": True,\n \"sanitize_session_files\": True,\n \"max_history_items\": 3,\n \"cache_dir\": \"~/.cache/cortex-skill\",\n \"cache_permissions\": \"0600\",\n \"allowed_envelopes\": [\"RO\", \"RW\", \"RESEARCH\"],\n \"deploy_envelope_confirmation\": True,\n \"execution_timeout_seconds\": 300,\n \"credential_file_allowlist\": [\n \"~/.ssh/*\",\n \"~/.snowflake/*\",\n \"**/.env\",\n \"**/.env.*\",\n \"**/credentials.json\",\n \"**/*_key.p8\",\n \"**/*_key.pem\",\n \"~/.aws/credentials\",\n \"~/.kube/config\"\n ]\n }\n }\n\n def __init__(\n self,\n config_path: Optional[Path] = None,\n org_policy_path: Optional[Path] = None\n ):\n \"\"\"Initialize config manager.\"\"\"\n self._config = self._load_config(config_path, org_policy_path)\n\n def _validate_config(self, config: Dict) -> None:\n \"\"\"Validate configuration values.\"\"\"\n security = config.get(\"security\", {})\n\n # Validate approval_mode\n approval_mode = security.get(\"approval_mode\")\n if approval_mode not in [\"prompt\", \"auto\", \"envelope_only\"]:\n raise ConfigValidationError(\n f\"Invalid approval_mode: {approval_mode}. \"\n f\"Must be one of: prompt, auto, envelope_only\"\n )\n\n # Validate allowed_envelopes\n valid_envelopes = {\"RO\", \"RW\", \"RESEARCH\", \"DEPLOY\", \"NONE\"}\n allowed_envelopes = security.get(\"allowed_envelopes\", [])\n for envelope in allowed_envelopes:\n if envelope not in valid_envelopes:\n raise ConfigValidationError(\n f\"Invalid envelope: {envelope}. \"\n f\"Must be one of: {', '.join(valid_envelopes)}\"\n )\n\n # Validate numeric values\n confidence = security.get(\"tool_prediction_confidence_threshold\")\n if confidence is not None:\n if not isinstance(confidence, (int, float)):\n raise ConfigValidationError(\n f\"tool_prediction_confidence_threshold must be a number, got {type(confidence).__name__}\"\n )\n if not (0 \u003c= confidence \u003c= 1):\n raise ConfigValidationError(\n f\"tool_prediction_confidence_threshold must be between 0 and 1, got {confidence}\"\n )\n\n retention = security.get(\"audit_log_retention\")\n if retention is not None:\n if not isinstance(retention, int):\n raise ConfigValidationError(\n f\"audit_log_retention must be an integer, got {type(retention).__name__}\"\n )\n if retention \u003c 0:\n raise ConfigValidationError(\n f\"audit_log_retention must be >= 0, got {retention}\"\n )\n\n def _safe_placeholder_path(self, original_path: str) -> str:\n \"\"\"Fallback when install-time __CODING_AGENT__ replacement was not applied.\"\"\"\n suffix = Path(original_path).name or \"audit.log\"\n return str(Path.home() / \".cache\" / \"cortex-skill\" / suffix)\n\n def _expand_paths(self, config: Dict) -> Dict:\n \"\"\"Expand ~ and environment variables in file paths.\"\"\"\n security = config.get(\"security\", {})\n\n # Expand audit_log_path\n if \"audit_log_path\" in security:\n security[\"audit_log_path\"] = os.path.expanduser(security[\"audit_log_path\"])\n if \"__CODING_AGENT__\" in security[\"audit_log_path\"]:\n security[\"audit_log_path\"] = self._safe_placeholder_path(security[\"audit_log_path\"])\n\n # Expand cache_dir\n if \"cache_dir\" in security:\n security[\"cache_dir\"] = os.path.expanduser(security[\"cache_dir\"])\n\n config[\"security\"] = security\n return config\n\n def _load_config(\n self,\n config_path: Optional[Path],\n org_policy_path: Optional[Path]\n ) -> Dict:\n \"\"\"Load configuration with 3-layer precedence.\"\"\"\n # Start with defaults\n config = copy.deepcopy(self.DEFAULT_CONFIG)\n\n # Load user config if exists\n if config_path and config_path.exists():\n try:\n with open(config_path, 'r') as f:\n try:\n user_config = yaml.safe_load(f) or {}\n config = self._merge_config(config, user_config)\n except yaml.YAMLError as e:\n print(f\"Warning: Failed to parse user config {config_path}: {e}\", file=sys.stderr)\n except OSError as e:\n print(f\"Warning: Failed to read user config {config_path}: {e}\", file=sys.stderr)\n\n org_policy_security = {}\n\n # Load org policy if exists\n if org_policy_path and org_policy_path.exists():\n try:\n with open(org_policy_path, 'r') as f:\n try:\n org_policy = yaml.safe_load(f) or {}\n org_policy_security = org_policy.get(\"security\", {}) or {}\n\n # If override flag set, org policy wins completely\n if org_policy.get(\"security\", {}).get(\"override_user_config\"):\n # Merge org policy over defaults (skip user config)\n config = self._merge_config(copy.deepcopy(self.DEFAULT_CONFIG), org_policy)\n else:\n # Normal merge: org policy > user config > defaults\n config = self._merge_config(config, org_policy)\n except yaml.YAMLError as e:\n print(f\"Warning: Failed to parse org policy {org_policy_path}: {e}\", file=sys.stderr)\n except OSError as e:\n print(f\"Warning: Failed to read org policy {org_policy_path}: {e}\", file=sys.stderr)\n\n # Validate before applying floors so invalid user config is still rejected.\n self._validate_config(config)\n\n # User config must not relax the security floor unless org policy\n # explicitly authorizes the relaxed field/value.\n config = self._enforce_security_floor(config, org_policy_security)\n\n # Validate configuration\n self._validate_config(config)\n\n # Expand file paths\n config = self._expand_paths(config)\n\n return config\n\n def _enforce_security_floor(self, config: Dict, org_policy_security: Optional[Dict] = None) -> Dict:\n \"\"\"Prevent user config from relaxing defaults without explicit org policy.\"\"\"\n result = copy.deepcopy(config)\n security = result.setdefault(\"security\", {})\n default_security = self.DEFAULT_CONFIG[\"security\"]\n org_policy_security = org_policy_security or {}\n\n if (\n security.get(\"approval_mode\") != default_security[\"approval_mode\"]\n and \"approval_mode\" not in org_policy_security\n ):\n security[\"approval_mode\"] = default_security[\"approval_mode\"]\n\n default_envelopes = set(default_security[\"allowed_envelopes\"])\n explicit_org_envelopes = set(org_policy_security.get(\"allowed_envelopes\", []))\n envelope_floor = default_envelopes | explicit_org_envelopes\n requested_envelopes = security.get(\"allowed_envelopes\", default_security[\"allowed_envelopes\"])\n security[\"allowed_envelopes\"] = [\n envelope for envelope in requested_envelopes\n if envelope in envelope_floor\n ]\n\n return result\n\n def _merge_config(self, base: Dict, override: Dict) -> Dict:\n \"\"\"Deep merge override into base.\"\"\"\n result = copy.deepcopy(base)\n for key, value in override.items():\n if key in result and isinstance(result[key], dict) and isinstance(value, dict):\n result[key] = self._merge_config(result[key], value)\n else:\n result[key] = value\n return result\n\n def get(self, key: str, default: Any = None) -> Any:\n \"\"\"Get config value by dot-notation key.\"\"\"\n keys = key.split(\".\")\n value = self._config\n\n for k in keys:\n if isinstance(value, dict) and k in value:\n value = value[k]\n else:\n return default\n\n return value\n","content_type":"text/x-python; charset=utf-8","language":"python","size":9141,"content_sha256":"b06fa8f7370d4bd73bae80370f34f0f64ea5aec5d7e2ebb6c4551dbd616e2cdd"},{"filename":"security/default_policy.yaml","content":"# Default security policy for cortex-code skill\n# This file documents the secure defaults - do not modify directly\n# To customize, create ~/.claude/skills/cortex-code/config.yaml\n\nsecurity:\n # Approval mode: \"prompt\" | \"auto\" | \"envelope_only\"\n # Default: \"prompt\" (most secure - ask user before execution)\n approval_mode: \"prompt\"\n\n # Tool prediction settings (for \"prompt\" mode)\n tool_prediction_confidence_threshold: 0.7\n allow_tool_expansion: true\n\n # Audit logging (mandatory when approval_mode: \"auto\")\n audit_log_path: \"~/.claude/skills/cortex-code/audit.log\"\n audit_log_rotation: \"10MB\"\n audit_log_retention: 30\n\n # Prompt sanitization\n sanitize_conversation_history: true\n sanitize_session_files: true\n max_history_items: 3\n\n # Cache security\n cache_dir: \"~/.cache/cortex-skill\"\n cache_permissions: \"0600\"\n\n # Envelope restrictions\n allowed_envelopes:\n - \"RO\"\n - \"RW\"\n - \"RESEARCH\"\n deploy_envelope_confirmation: true\n\n # Routing security - never route these to Cortex\n credential_file_allowlist:\n - \"~/.ssh/*\"\n - \"~/.snowflake/*\"\n - \"**/.env\"\n - \"**/.env.*\"\n - \"**/credentials.json\"\n - \"**/*_key.p8\"\n - \"**/*_key.pem\"\n - \"~/.aws/credentials\"\n - \"~/.kube/config\"\n","content_type":"application/yaml; charset=utf-8","language":"yaml","size":1230,"content_sha256":"6bd6b5528837590928eac6c96126dd40e8cef2c13e2e467ac57eb5904cdff981"},{"filename":"security/policies/default_policy.yaml","content":"# Default security policy for cortex-code skill\n# This file documents the secure defaults - do not modify directly\n# To customize, create ~/.claude/skills/cortex-code/config.yaml\n\nsecurity:\n # Approval mode: \"prompt\" | \"auto\" | \"envelope_only\"\n # Default: \"prompt\" (most secure - ask user before execution)\n approval_mode: \"prompt\"\n\n # Tool prediction settings (for \"prompt\" mode)\n tool_prediction_confidence_threshold: 0.7\n allow_tool_expansion: true\n\n # Audit logging (mandatory when approval_mode: \"auto\")\n audit_log_path: \"~/.claude/skills/cortex-code/audit.log\"\n audit_log_rotation: \"10MB\"\n audit_log_retention: 30\n\n # Prompt sanitization\n sanitize_conversation_history: true\n sanitize_session_files: true\n max_history_items: 3\n\n # Cache security\n cache_dir: \"~/.cache/cortex-skill\"\n cache_permissions: \"0600\"\n\n # Envelope restrictions\n allowed_envelopes:\n - \"RO\"\n - \"RW\"\n - \"RESEARCH\"\n deploy_envelope_confirmation: true\n\n # Routing security - never route these to Cortex\n credential_file_allowlist:\n - \"~/.ssh/*\"\n - \"~/.snowflake/*\"\n - \"**/.env\"\n - \"**/.env.*\"\n - \"**/credentials.json\"\n - \"**/*_key.p8\"\n - \"**/*_key.pem\"\n - \"~/.aws/credentials\"\n - \"~/.kube/config\"\n","content_type":"application/yaml; charset=utf-8","language":"yaml","size":1230,"content_sha256":"6bd6b5528837590928eac6c96126dd40e8cef2c13e2e467ac57eb5904cdff981"},{"filename":"security/prompt_sanitizer.py","content":"\"\"\"Prompt sanitizer for PII removal and injection detection.\"\"\"\n\nimport re\nimport unicodedata\nfrom typing import List, Dict, Any\n\n\nclass PromptSanitizer:\n \"\"\"Sanitizes prompts by removing PII and detecting injection attempts.\"\"\"\n\n # PII regex patterns\n CREDIT_CARD_PATTERN = re.compile(\n r'\\b(?:\\d{4}[-\\s]?){3}\\d{4}\\b' # Matches formats: 1234-5678-9012-3456 or 1234567890123456\n )\n\n SSN_PATTERN = re.compile(\n r'\\b\\d{3}-\\d{2}-\\d{4}\\b|' # Matches: 123-45-6789\n r'\\b\\d{9}\\b' # Matches: 123456789 (exactly 9 digits)\n )\n\n EMAIL_PATTERN = re.compile(\n r'\\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Z|a-z]{2,}\\b'\n )\n\n PHONE_PATTERN = re.compile(\n r'\\b(?:\\+?1[-.\\s]?)?\$?\\d{3}\$?[-.\\s]?\\d{3}[-.\\s]?\\d{4}\\b'\n )\n\n API_KEY_PATTERN = re.compile(\n r'\\b(?:api[_-]?key|token|secret)\\s*[:=]\\s*[\"\\']?[A-Za-z0-9_./+=-]{8,}[\"\\']?|'\n r'\\bsk-[A-Za-z0-9_./+=-]{8,}\\b|'\n r'\\b[A-Za-z0-9]{32,}\\b',\n re.IGNORECASE,\n )\n\n ZERO_WIDTH_PATTERN = re.compile(r'[\\u200B-\\u200D\\uFEFF]')\n HOMOGLYPH_TRANSLATION = str.maketrans({\n 'а': 'a', 'А': 'A', # Cyrillic a\n 'е': 'e', 'Е': 'E', # Cyrillic e\n 'і': 'i', 'І': 'I', # Cyrillic/Ukrainian i\n 'о': 'o', 'О': 'O', # Cyrillic o\n 'р': 'p', 'Р': 'P', # Cyrillic er\n 'с': 'c', 'С': 'C', # Cyrillic es\n 'х': 'x', 'Х': 'X', # Cyrillic ha\n 'у': 'y', 'У': 'Y', # Cyrillic u\n })\n\n # Injection detection patterns\n INJECTION_PATTERNS = [\n re.compile(r'ignore\\s+(?:all\\s+|the\\s+)?(previous|above|prior)\\s+(instructions|directions|prompts?)', re.IGNORECASE),\n re.compile(r'(enter|enable|activate)\\s+developer\\s+mode', re.IGNORECASE),\n re.compile(r'you\\s+are\\s+now\\s+in\\s+developer\\s+mode', re.IGNORECASE),\n re.compile(r'disregard\\s+(?:all\\s+|the\\s+)?(previous|above|prior)', re.IGNORECASE),\n re.compile(r'bypass\\s+(restrictions|rules|guidelines)', re.IGNORECASE),\n ]\n\n def _normalize_for_detection(self, text: str) -> str:\n \"\"\"Normalize text so obfuscated prompt injections match detection rules.\"\"\"\n normalized = unicodedata.normalize('NFKC', text)\n normalized = self.ZERO_WIDTH_PATTERN.sub('', normalized)\n normalized = normalized.translate(self.HOMOGLYPH_TRANSLATION)\n normalized = ''.join(\n char for char in normalized\n if unicodedata.category(char) not in {'Cf', 'Mn'}\n )\n return normalized\n\n def sanitize(self, text: str) -> str:\n \"\"\"\n Sanitize text by removing PII and detecting injection attempts.\n\n Args:\n text: The text to sanitize\n\n Returns:\n Sanitized text with PII removed and injection warnings added\n \"\"\"\n if not text:\n return text\n\n detection_text = self._normalize_for_detection(text)\n\n # Check for injection attempts first\n for pattern in self.INJECTION_PATTERNS:\n if pattern.search(detection_text):\n return \"[POTENTIAL INJECTION DETECTED - REMOVED]\"\n\n # Remove PII\n text = self.CREDIT_CARD_PATTERN.sub('\u003cCREDIT_CARD>', text)\n text = self.SSN_PATTERN.sub('\u003cSSN>', text)\n text = self.EMAIL_PATTERN.sub('\u003cEMAIL>', text)\n text = self.PHONE_PATTERN.sub('\u003cPHONE>', text)\n text = self.API_KEY_PATTERN.sub('[API_KEY_REDACTED]', text)\n\n return text\n\n def sanitize_sql_literals(self, sql: str) -> str:\n \"\"\"\n Sanitize SQL string by removing PII from literals.\n\n Args:\n sql: The SQL string to sanitize\n\n Returns:\n Sanitized SQL string\n \"\"\"\n return self.sanitize(sql)\n\n def sanitize_history(self, history: List[Dict[str, Any]], max_items: int = 3) -> List[Dict[str, Any]]:\n \"\"\"\n Sanitize conversation history by limiting items and removing PII.\n\n Args:\n history: List of conversation history items (dicts with 'role' and 'content')\n max_items: Maximum number of items to keep (default: 3)\n\n Returns:\n Sanitized and limited history list\n \"\"\"\n if not history:\n return []\n\n # Keep only the last max_items\n limited_history = history[-max_items:] if len(history) > max_items else history\n\n # Sanitize each item's content\n sanitized = []\n for item in limited_history:\n sanitized_item = item.copy()\n if 'content' in sanitized_item:\n sanitized_item['content'] = self.sanitize(sanitized_item['content'])\n sanitized.append(sanitized_item)\n\n return sanitized\n","content_type":"text/x-python; charset=utf-8","language":"python","size":4678,"content_sha256":"12b7b3f2daf64bb3a3f3413f9fab65447ec734e24c026a2d68e2a5c489d8bcd5"}],"content_json":{"type":"doc","content":[{"type":"heading","attrs":{"level":1},"content":[{"text":"Cortex Code Integration Skill","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Install","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"# Install via npm skills ecosystem (works with Claude Code, Cursor, Codex, and 40+ agents)\nnpx skills add snowflake-labs/subagent-cortex-code --copy\n\n# Prerequisite: Cortex Code CLI must be installed and configured\n# See: https://docs.snowflake.com/en/user-guide/cortex-code\nwhich cortex # verify installation","type":"text"}]},{"type":"paragraph","content":[{"text":"This skill enables your coding agent to leverage Cortex Code's specialized Snowflake expertise by intelligently routing Snowflake-related operations to Cortex Code CLI in headless mode.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Architecture Overview","type":"text"}]},{"type":"paragraph","content":[{"text":"Routing Principle","type":"text","marks":[{"type":"strong"}]},{"text":": ONLY Snowflake operations → Cortex Code. Everything else → your coding agent.","type":"text"}]},{"type":"paragraph","content":[{"text":"Key Components","type":"text","marks":[{"type":"strong"}]},{"text":":","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Dynamic skill discovery at session initialization","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"LLM-based semantic routing (not keyword matching)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Security wrapper with approval modes (prompt/auto/envelope_only)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Stateless Cortex execution with context enrichment","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Hybrid memory management","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Audit logging for compliance","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Security","type":"text"}]},{"type":"paragraph","content":[{"text":"The skill includes a security wrapper around Cortex execution with three approval modes:","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Approval Modes","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"prompt","type":"text","marks":[{"type":"strong"}]},{"text":" (default): High security","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"User shown approval prompt with predicted tools and confidence","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"User must approve before execution","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"No audit logging required","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Best for: Interactive sessions, untrusted prompts, production","type":"text"}]}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"auto","type":"text","marks":[{"type":"strong"}]},{"text":": Medium security","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"All operations auto-approved","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Mandatory audit logging","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Envelopes still enforced","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Best for: Automated workflows, trusted environments","type":"text"}]}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"envelope_only","type":"text","marks":[{"type":"strong"}]},{"text":": Medium security","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"No tool prediction (faster)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Auto-approved with audit logging","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Relies on envelope blocklist only","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Best for: Trusted environments, low latency needs","type":"text"}]}]}]}]}]},{"type":"paragraph","content":[{"text":"Configuration","type":"text","marks":[{"type":"strong"}]},{"text":": Set in ","type":"text"},{"text":"config.yaml","type":"text","marks":[{"type":"code_inline"}]},{"text":" in the skill's install directory, or via organization policy.","type":"text"}]},{"type":"blockquote","content":[{"type":"paragraph","content":[{"text":"IMPORTANT — ","type":"text","marks":[{"type":"strong"}]},{"text":"config.yaml","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" is optional.","type":"text","marks":[{"type":"strong"}]},{"text":" The skill ships only ","type":"text"},{"text":"config.yaml.example","type":"text","marks":[{"type":"code_inline"}]},{"text":" as a template. If no ","type":"text"},{"text":"config.yaml","type":"text","marks":[{"type":"code_inline"}]},{"text":" exists, the Python scripts apply safe defaults (","type":"text"},{"text":"approval_mode: prompt","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"default_envelope: RO","type":"text","marks":[{"type":"code_inline"}]},{"text":"). ","type":"text"},{"text":"Do not search, glob, or ","type":"text","marks":[{"type":"strong"}]},{"text":"ls","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" for ","type":"text","marks":[{"type":"strong"}]},{"text":"config.yaml","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" before executing","type":"text","marks":[{"type":"strong"}]},{"text":" — ","type":"text"},{"text":"ConfigManager","type":"text","marks":[{"type":"code_inline"}]},{"text":" handles this internally. Only read/create ","type":"text"},{"text":"config.yaml","type":"text","marks":[{"type":"code_inline"}]},{"text":" if the user explicitly asks to change settings.","type":"text"}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Built-in Protections","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Prompt Sanitization","type":"text","marks":[{"type":"strong"}]},{"text":": Automatic PII removal and injection detection","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Credential Blocking","type":"text","marks":[{"type":"strong"}]},{"text":": Prevents routing when credential paths detected","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Secure Caching","type":"text","marks":[{"type":"strong"}]},{"text":": SHA256-validated cache in ","type":"text"},{"text":"~/.cache/cortex-skill/","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Audit Logging","type":"text","marks":[{"type":"strong"}]},{"text":": Structured JSONL logs (mandatory for auto/envelope_only)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Organization Policy","type":"text","marks":[{"type":"strong"}]},{"text":": Enterprise override via ","type":"text"},{"text":"~/.snowflake/cortex/claude-skill-policy.yaml","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Fast Path for Repeat Queries","type":"text"}]},{"type":"paragraph","content":[{"text":"Session state is cached — do not re-run initialization steps on every query.","type":"text","marks":[{"type":"strong"}]}]},{"type":"paragraph","content":[{"text":"Skip the following steps if they've already run in the current session:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"discover_cortex.py","type":"text","marks":[{"type":"code_inline"}]},{"text":" — output cached to ","type":"text"},{"text":"~/.cache/cortex-skill/cortex-capabilities.json","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"route_request.py","type":"text","marks":[{"type":"code_inline"}]},{"text":" — for obvious Snowflake queries (user says \"Snowflake\", \"Cortex\", \"databases\", \"warehouse\", etc.), you can skip routing and go straight to execution","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"cortex connections list","type":"text","marks":[{"type":"code_inline"}]},{"text":" — the active connection doesn't change within a session; reuse it","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Any ","type":"text"},{"text":"config.yaml","type":"text","marks":[{"type":"code_inline"}]},{"text":" / org-policy inspection — ","type":"text"},{"text":"ConfigManager","type":"text","marks":[{"type":"code_inline"}]},{"text":" handles this (see note above)","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Minimal flow for a follow-up Snowflake query","type":"text","marks":[{"type":"strong"}]},{"text":" (after the first query in a session):","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"(If ","type":"text"},{"text":"approval_mode: prompt","type":"text","marks":[{"type":"code_inline"}]},{"text":") ask user for approval","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Call ","type":"text"},{"text":"execute_cortex.py","type":"text","marks":[{"type":"code_inline"}]},{"text":" with the enriched prompt and envelope","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Return results","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"That's it. Three steps — no re-discovery, no re-routing, no config inspection.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Session Initialization","type":"text"}]},{"type":"paragraph","content":[{"text":"When this skill is first loaded:","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Step 1: Discover Cortex Capabilities","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"PYTHON=$(command -v python3 2>/dev/null || command -v python 2>/dev/null || echo python3)\n$PYTHON scripts/discover_cortex.py","type":"text"}]},{"type":"paragraph","content":[{"text":"This script:","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Runs ","type":"text"},{"text":"cortex skill list","type":"text","marks":[{"type":"code_inline"}]},{"text":" to enumerate all available Cortex skills","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Reads each skill's SKILL.md frontmatter and trigger patterns","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Caches capabilities with ","type":"text"},{"text":"CacheManager","type":"text","marks":[{"type":"code_inline"}]},{"text":" in the configured cache directory","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Returns structured data about what Cortex can handle","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Expected output: JSON mapping of skill names to their trigger patterns and capabilities.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Step 2: Load Routing Context","type":"text"}]},{"type":"paragraph","content":[{"text":"The discovered capabilities are loaded into memory to inform routing decisions throughout the session.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Workflow: Handling User Requests","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Step 1: Analyze Request with LLM-Based Routing","type":"text"}]},{"type":"paragraph","content":[{"text":"Before taking any action, analyze the user's request:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"PYTHON=$(command -v python3 2>/dev/null || command -v python 2>/dev/null || echo python3)\n$PYTHON scripts/route_request.py --prompt \"USER_PROMPT_HERE\"","type":"text"}]},{"type":"paragraph","content":[{"text":"This script:","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Loads Cortex capabilities from cache","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Uses LLM reasoning to classify the request","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Returns routing decision with confidence score","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Routing Logic","type":"text","marks":[{"type":"strong"}]},{"text":":","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Route to Cortex","type":"text","marks":[{"type":"strong"}]},{"text":" if request involves:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Snowflake databases, warehouses, schemas, tables","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"SQL queries specifically for Snowflake","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Cortex AI features (Cortex Search, Cortex Analyst, ML functions)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Snowpark, dynamic tables, streams, tasks","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Data governance, data quality, or security in Snowflake context","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"User explicitly mentions \"Cortex\" or \"Snowflake\"","type":"text"}]}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Route to your coding agent","type":"text","marks":[{"type":"strong"}]},{"text":" if request involves:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Local file operations (reading, writing, editing local files)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"General programming (Python, JavaScript, etc. not Snowflake-specific)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Non-Snowflake databases (PostgreSQL, MySQL, MongoDB, etc.)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Web development, frontend work","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Infrastructure/DevOps unrelated to Snowflake","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Git operations, GitHub, version control","type":"text"}]}]}]}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Step 2: Execute Based on Routing Decision","type":"text"}]},{"type":"heading","attrs":{"level":4},"content":[{"text":"If routing is ","type":"text"},{"text":"coding_agent","type":"text","marks":[{"type":"code_inline"}]},{"text":" (handle locally):","type":"text"}]},{"type":"paragraph","content":[{"text":"Handle the request directly using your agent's built-in capabilities. No Cortex involvement.","type":"text"}]},{"type":"heading","attrs":{"level":4},"content":[{"text":"If routed to Cortex Code:","type":"text"}]},{"type":"paragraph","content":[{"text":"Proceed to Step 3.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Step 3: Choose Security Envelope and Handle Approval","type":"text"}]},{"type":"paragraph","content":[{"text":"Before executing Cortex, the security wrapper handles approval based on configured mode.","type":"text"}]},{"type":"heading","attrs":{"level":4},"content":[{"text":"Step 3a: Check Approval Mode","type":"text"}]},{"type":"paragraph","content":[{"text":"security_wrapper.py","type":"text","marks":[{"type":"code_inline"}]},{"text":" reads ","type":"text"},{"text":"approval_mode","type":"text","marks":[{"type":"code_inline"}]},{"text":" from ","type":"text"},{"text":"config.yaml","type":"text","marks":[{"type":"code_inline"}]},{"text":" internally — ","type":"text"},{"text":"do not inspect the config file yourself.","type":"text","marks":[{"type":"strong"}]},{"text":" If ","type":"text"},{"text":"config.yaml","type":"text","marks":[{"type":"code_inline"}]},{"text":" doesn't exist, the default is ","type":"text"},{"text":"prompt","type":"text","marks":[{"type":"code_inline"}]},{"text":" mode.","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"prompt mode","type":"text","marks":[{"type":"strong"}]},{"text":" (default): Requires user approval","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"auto mode","type":"text","marks":[{"type":"strong"}]},{"text":": Auto-approve with audit logging","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"envelope_only mode","type":"text","marks":[{"type":"strong"}]},{"text":": Auto-approve, no tool prediction","type":"text"}]}]}]},{"type":"heading","attrs":{"level":4},"content":[{"text":"Step 3b: Handle Approval (if prompt mode)","type":"text"}]},{"type":"paragraph","content":[{"text":"If using prompt mode:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"PYTHON=$(command -v python3 2>/dev/null || command -v python 2>/dev/null || echo python3)\n$PYTHON scripts/security_wrapper.py \\\n --prompt \"ENRICHED_PROMPT\" \\\n --envelope \"RW\"","type":"text"}]},{"type":"paragraph","content":[{"text":"This will:","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Predict required tools using LLM","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Display approval prompt to user:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":""},"content":[{"text":"Cortex Code needs to execute the following tools:\n\n • snowflake_sql_execute\n • Read\n • Write\n\nEnvelope: RW\nConfidence: 85%\n\nApprove execution? [yes/no]","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"If approved, proceed to Step 3c","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"If denied, abort execution","type":"text"}]}]}]},{"type":"heading","attrs":{"level":4},"content":[{"text":"Step 3c: Determine Security Envelope","type":"text"}]},{"type":"paragraph","content":[{"text":"Determine the appropriate security envelope based on the operation:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"RO","type":"text","marks":[{"type":"strong"}]},{"text":" (Read-Only): For queries and read operations - blocks Edit, Write, destructive Bash","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"RW","type":"text","marks":[{"type":"strong"}]},{"text":" (Read-Write): For data modifications - allows most operations, blocks destructive Bash","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"RESEARCH","type":"text","marks":[{"type":"strong"}]},{"text":": For exploratory work - read access plus web tools","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"DEPLOY","type":"text","marks":[{"type":"strong"}]},{"text":": For deployment operations - blocks destructive Bash commands","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"NONE","type":"text","marks":[{"type":"strong"}]},{"text":": Custom blocklist via --disallowed-tools","type":"text"}]}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Step 4: Enrich Context for Cortex","type":"text"}]},{"type":"paragraph","content":[{"text":"Build an enriched prompt that includes:","type":"text"}]},{"type":"paragraph","content":[{"text":"Claude Conversation Context","type":"text","marks":[{"type":"strong"}]},{"text":":","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Last 2-3 relevant exchanges from current Claude session","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Any Snowflake-specific details already discussed","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Recent Cortex Session Context","type":"text","marks":[{"type":"strong"}]},{"text":":","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"PYTHON=$(command -v python3 2>/dev/null || command -v python 2>/dev/null || echo python3)\n$PYTHON scripts/read_cortex_sessions.py --limit 3","type":"text"}]},{"type":"paragraph","content":[{"text":"This reads the most recent Cortex session files from ","type":"text"},{"text":"~/.local/share/cortex/sessions/","type":"text","marks":[{"type":"code_inline"}]},{"text":" to understand what Cortex recently worked on.","type":"text"}]},{"type":"paragraph","content":[{"text":"Enriched Prompt Format","type":"text","marks":[{"type":"strong"}]},{"text":":","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":""},"content":[{"text":"# Context from Current Session\n[Recent relevant conversation history]\n\n# Recent Cortex Work\n[Summary from recent Cortex sessions]\n\n# User Request\n[Original user prompt]","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Step 5: Execute Cortex Code Headlessly","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"PYTHON=$(command -v python3 2>/dev/null || command -v python 2>/dev/null || echo python3)\n$PYTHON scripts/execute_cortex.py \\\n --prompt \"ENRICHED_PROMPT\" \\\n --connection \"connection_name\" \\\n --envelope \"RW\" \\\n --disallowed-tools \"tool1\" \"tool2\"","type":"text"}]},{"type":"paragraph","content":[{"text":"This script:","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Invokes ","type":"text"},{"text":"cortex -p \"prompt\" --output-format stream-json","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Uses print mode for prompt delivery and stream JSON output for non-TTY parsing","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Applies envelope-based security via ","type":"text"},{"text":"--disallowed-tools","type":"text","marks":[{"type":"code_inline"}]},{"text":" blocklist for safety","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Parses NDJSON event stream in real-time","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Detects tool use events and execution results","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Key Insight","type":"text","marks":[{"type":"strong"}]},{"text":": The wrapper intentionally does not combine ","type":"text"},{"text":"-p","type":"text","marks":[{"type":"code_inline"}]},{"text":" with ","type":"text"},{"text":"--input-format stream-json","type":"text","marks":[{"type":"code_inline"}]},{"text":". Cortex reserves ","type":"text"},{"text":"--input-format","type":"text","marks":[{"type":"code_inline"}]},{"text":" for JSON stdin input; with closed stdin, that combination can emit only an init event and exit before processing the prompt.","type":"text"}]},{"type":"paragraph","content":[{"text":"Security Envelopes","type":"text","marks":[{"type":"strong"}]},{"text":":","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"RO","type":"text","marks":[{"type":"strong"}]},{"text":" (Read-Only): Blocks Edit, Write, destructive Bash commands","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"RW","type":"text","marks":[{"type":"strong"}]},{"text":" (Read-Write): Blocks destructive operations like rm -rf, sudo","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"RESEARCH","type":"text","marks":[{"type":"strong"}]},{"text":": Read access plus web tools, blocks write operations","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"DEPLOY","type":"text","marks":[{"type":"strong"}]},{"text":": Deployment operations, blocks destructive Bash commands","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"NONE","type":"text","marks":[{"type":"strong"}]},{"text":": Custom blocklist via --disallowed-tools parameter","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Event Stream Handling","type":"text","marks":[{"type":"strong"}]},{"text":":","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"type: assistant","type":"text","marks":[{"type":"code_inline"}]},{"text":" → Cortex's responses, display to user","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"type: tool_use","type":"text","marks":[{"type":"code_inline"}]},{"text":" → Cortex is calling a tool","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"type: result","type":"text","marks":[{"type":"code_inline"}]},{"text":" → Final outcome","type":"text"}]}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Step 6: Handle Permission Requests","type":"text"}]},{"type":"paragraph","content":[{"text":"With the security wrapper:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"prompt mode","type":"text","marks":[{"type":"strong"}]},{"text":": User approves BEFORE execution (no mid-execution prompts)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"auto/envelope_only modes","type":"text","marks":[{"type":"strong"}]},{"text":": Non-blocked tools are auto-approved in stream JSON mode","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"The security wrapper handles permission management through:","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Upfront approval","type":"text","marks":[{"type":"strong"}]},{"text":" (prompt mode): User approves predicted tools before execution","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Audit logging","type":"text","marks":[{"type":"strong"}]},{"text":" (auto/envelope_only): All operations logged to ","type":"text"},{"text":"audit.log","type":"text","marks":[{"type":"code_inline"}]},{"text":" in the skill's install directory","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Envelope enforcement","type":"text","marks":[{"type":"strong"}]},{"text":": Tool blocklist still enforced via ","type":"text"},{"text":"--disallowed-tools","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Step 7: Return Results to User","type":"text"}]},{"type":"paragraph","content":[{"text":"Format Cortex's output for the current session:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Show SQL query results in readable format","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Display any generated artifacts","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Report success/failure status","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Provide relevant excerpts from Cortex's analysis","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Examples","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Example 1: Snowflake Query","type":"text"}]},{"type":"paragraph","content":[{"text":"User says","type":"text","marks":[{"type":"strong"}]},{"text":": \"Show me the top 10 customers by revenue in Snowflake\"","type":"text"}]},{"type":"paragraph","content":[{"text":"Routing","type":"text","marks":[{"type":"strong"}]},{"text":": → Cortex Code (Snowflake SQL query)","type":"text"}]},{"type":"paragraph","content":[{"text":"Security Envelope","type":"text","marks":[{"type":"strong"}]},{"text":": RW (allows SQL execution)","type":"text"}]},{"type":"paragraph","content":[{"text":"Cortex Action","type":"text","marks":[{"type":"strong"}]},{"text":":","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Uses snowflake_sql_execute to run: ","type":"text"},{"text":"SELECT customer_name, SUM(revenue) as total FROM sales GROUP BY customer_name ORDER BY total DESC LIMIT 10","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Returns formatted results","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Result","type":"text","marks":[{"type":"strong"}]},{"text":": Table displayed to user with top 10 customers.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Example 2: Local File Operation","type":"text"}]},{"type":"paragraph","content":[{"text":"User says","type":"text","marks":[{"type":"strong"}]},{"text":": \"Read the config.json file in this directory\"","type":"text"}]},{"type":"paragraph","content":[{"text":"Routing","type":"text","marks":[{"type":"strong"}]},{"text":": → your coding agent (local file operation)","type":"text"}]},{"type":"paragraph","content":[{"text":"Claude Action","type":"text","marks":[{"type":"strong"}]},{"text":": Uses Read tool directly, no Cortex involvement.","type":"text"}]},{"type":"paragraph","content":[{"text":"Result","type":"text","marks":[{"type":"strong"}]},{"text":": File contents displayed.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Example 3: Data Quality Check","type":"text"}]},{"type":"paragraph","content":[{"text":"User says","type":"text","marks":[{"type":"strong"}]},{"text":": \"Check data quality for the SALES_DATA table\"","type":"text"}]},{"type":"paragraph","content":[{"text":"Routing","type":"text","marks":[{"type":"strong"}]},{"text":": → Cortex Code (Snowflake data quality - matches Cortex's data-quality skill)","type":"text"}]},{"type":"paragraph","content":[{"text":"Security Envelope","type":"text","marks":[{"type":"strong"}]},{"text":": RW (allows SQL execution for analysis)","type":"text"}]},{"type":"paragraph","content":[{"text":"Cortex Action","type":"text","marks":[{"type":"strong"}]},{"text":":","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Runs data quality checks using its data-quality skill","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Analyzes schema, null rates, duplicates, etc.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Generates quality report","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Result","type":"text","marks":[{"type":"strong"}]},{"text":": Comprehensive data quality report with recommendations.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Important Notes","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Security Wrapper","type":"text"}]},{"type":"paragraph","content":[{"text":"The skill uses a security wrapper that provides:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Approval modes","type":"text","marks":[{"type":"strong"}]},{"text":": prompt (default), auto, envelope_only","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Prompt sanitization","type":"text","marks":[{"type":"strong"}]},{"text":": Automatic PII removal and injection detection","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Credential blocking","type":"text","marks":[{"type":"strong"}]},{"text":": Prevents routing when credential paths detected","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Audit logging","type":"text","marks":[{"type":"strong"}]},{"text":": Mandatory for auto/envelope_only modes","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Tool prediction","type":"text","marks":[{"type":"strong"}]},{"text":": LLM predicts required tools for approval prompt","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Configuration","type":"text","marks":[{"type":"strong"}]},{"text":": ","type":"text"},{"text":"config.yaml","type":"text","marks":[{"type":"code_inline"}]},{"text":" in the skill's install directory, or via organization policy","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Headless Execution with Auto-Approval","type":"text"}]},{"type":"paragraph","content":[{"text":"When using auto or envelope_only modes:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"All tool calls are automatically approved without interactive prompts","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Works for built-in tools (Read, Write, Edit, Bash, Grep, Glob) and non-builtin tools (snowflake_sql_execute, data_diff, MCP tools)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Uses print mode for prompt delivery and stream JSON mode for non-TTY output parsing","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Security is controlled via ","type":"text"},{"text":"--disallowed-tools","type":"text","marks":[{"type":"code_inline"}]},{"text":" blocklist instead of interactive approval; use these modes only in trusted contexts","type":"text"}]}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Stateless Execution","type":"text"}]},{"type":"paragraph","content":[{"text":"Each Cortex invocation is stateless. Context must be explicitly provided via enriched prompts.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Memory Boundaries","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Your coding agent maintains","type":"text","marks":[{"type":"strong"}]},{"text":": Full conversation history, user preferences, project context","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Cortex Code receives","type":"text","marks":[{"type":"strong"}]},{"text":": Only task-specific context for current operation","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Cortex sessions are read","type":"text","marks":[{"type":"strong"}]},{"text":": For historical context enrichment only","type":"text"}]}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Security Envelope Strategy","type":"text"}]},{"type":"paragraph","content":[{"text":"Choose envelopes based on operation risk:","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Start with RO or RW","type":"text","marks":[{"type":"strong"}]},{"text":": Most operations fit here","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Use RESEARCH","type":"text","marks":[{"type":"strong"}]},{"text":": When web access is needed for exploratory work","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Use DEPLOY","type":"text","marks":[{"type":"strong"}]},{"text":": Only for deployment-style operations that require broader non-destructive tool access","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Use NONE with custom blocklist","type":"text","marks":[{"type":"strong"}]},{"text":": When fine-grained control is needed","type":"text"}]}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Performance Considerations","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Cortex skill discovery runs once per session (cached)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Each Cortex execution adds ~2-5 seconds latency","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Use routing wisely to minimize unnecessary Cortex calls","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Troubleshooting","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Error: \"Cortex CLI not found\"","type":"text"}]},{"type":"paragraph","content":[{"text":"Cause","type":"text","marks":[{"type":"strong"}]},{"text":": Cortex Code is not installed or not in PATH","type":"text"}]},{"type":"paragraph","content":[{"text":"Solution","type":"text","marks":[{"type":"strong"}]},{"text":":","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"which cortex\n# If not found, check installation: ~/.snowflake/cortex/","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Error: Approval prompt not appearing (or appearing unexpectedly)","type":"text"}]},{"type":"paragraph","content":[{"text":"Cause","type":"text","marks":[{"type":"strong"}]},{"text":": Approval mode misconfiguration or organization policy override","type":"text"}]},{"type":"paragraph","content":[{"text":"Solution","type":"text","marks":[{"type":"strong"}]},{"text":":","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"# Check approval mode (path varies by agent: ~/.claude/, ~/.cursor/, ~/.codex/, etc.)\ncat \"$(dirname $(which cortex))/../skills/cortex-code/config.yaml\" | grep approval_mode 2>/dev/null \\\n || cat ~/skills/cortex-code/config.yaml | grep approval_mode\n\n# Check organization policy (overrides user config)\ncat ~/.snowflake/cortex/claude-skill-policy.yaml 2>/dev/null\n\n# Expected:\n# prompt = shows approval prompts (default)\n# auto = auto-approves all operations\n# envelope_only = auto-approves, no tool prediction","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Error: \"Prompt contains credential file path\"","type":"text"}]},{"type":"paragraph","content":[{"text":"Cause","type":"text","marks":[{"type":"strong"}]},{"text":": Prompt mentions paths matching credential allowlist (e.g., ~/.ssh/, .env)","type":"text"}]},{"type":"paragraph","content":[{"text":"Solution","type":"text","marks":[{"type":"strong"}]},{"text":":","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Remove credential references from prompt","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Or customize allowlist in config.yaml if false positive","type":"text"}]}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Error: PII removed from prompts","type":"text"}]},{"type":"paragraph","content":[{"text":"Symptom","type":"text","marks":[{"type":"strong"}]},{"text":": Emails, phone numbers replaced with placeholders","type":"text"}]},{"type":"paragraph","content":[{"text":"Cause","type":"text","marks":[{"type":"strong"}]},{"text":": Automatic sanitization enabled by default","type":"text"}]},{"type":"paragraph","content":[{"text":"Solution","type":"text","marks":[{"type":"strong"}]},{"text":": Disable if needed (not recommended):","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"yaml"},"content":[{"text":"security:\n sanitize_conversation_history: false","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Error: \"Permission denied\" despite auto mode","type":"text"}]},{"type":"paragraph","content":[{"text":"Cause","type":"text","marks":[{"type":"strong"}]},{"text":": Tool is in the --disallowed-tools blocklist for current envelope","type":"text"}]},{"type":"paragraph","content":[{"text":"Solution","type":"text","marks":[{"type":"strong"}]},{"text":":","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Check which envelope is being used (RO/RW/RESEARCH/DEPLOY)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"If operation is safe, switch to a less restrictive envelope","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Avoid ","type":"text"},{"text":"NONE","type":"text","marks":[{"type":"code_inline"}]},{"text":" in auto/envelope_only modes; use a named envelope plus explicit custom blocklist if needed","type":"text"}]}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Error: Audit log not created","type":"text"}]},{"type":"paragraph","content":[{"text":"Symptom","type":"text","marks":[{"type":"strong"}]},{"text":": No audit.log despite auto/envelope_only mode","type":"text"}]},{"type":"paragraph","content":[{"text":"Solution","type":"text","marks":[{"type":"strong"}]},{"text":":","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"# Create the skill's install directory if missing and set permissions\n# Path is agent-specific: ~/.claude/skills/cortex-code/, ~/.cursor/skills/cortex-code/, etc.\nchmod 700 \"$(cd \"$(dirname \"$0\")/..\" && pwd)\"\n\n# Verify audit_log_path in config.yaml within the skill directory\ngrep audit_log_path config.yaml","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Error: Tools still requiring approval","type":"text"}]},{"type":"paragraph","content":[{"text":"Cause","type":"text","marks":[{"type":"strong"}]},{"text":": Approval mode, envelope blocklist, or stream JSON invocation is misconfigured","type":"text"}]},{"type":"paragraph","content":[{"text":"Solution","type":"text","marks":[{"type":"strong"}]},{"text":": Ensure the wrapper invokes ","type":"text"},{"text":"cortex -p \"...\" --output-format stream-json","type":"text","marks":[{"type":"code_inline"}]},{"text":" without ","type":"text"},{"text":"--input-format","type":"text","marks":[{"type":"code_inline"}]},{"text":", and that the configured envelope does not block the intended tool.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Issue: Routing sends Snowflake query to your coding agent","type":"text"}]},{"type":"paragraph","content":[{"text":"Cause","type":"text","marks":[{"type":"strong"}]},{"text":": Routing logic didn't detect Snowflake keywords","type":"text"}]},{"type":"paragraph","content":[{"text":"Solution","type":"text","marks":[{"type":"strong"}]},{"text":":","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Check if user mentioned \"Snowflake\" explicitly","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Review routing script logic in ","type":"text"},{"text":"scripts/route_request.py","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Add more trigger patterns to routing context","type":"text"}]}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Issue: Cortex returns \"Connection refused\"","type":"text"}]},{"type":"paragraph","content":[{"text":"Cause","type":"text","marks":[{"type":"strong"}]},{"text":": Snowflake connection not configured in Cortex","type":"text"}]},{"type":"paragraph","content":[{"text":"Solution","type":"text","marks":[{"type":"strong"}]},{"text":":","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"bash"},"content":[{"text":"cortex connections list\n# Verify connection is active\n# Check ~/.snowflake/cortex/settings.json for cortexAgentConnectionName","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Issue: Context enrichment too large","type":"text"}]},{"type":"paragraph","content":[{"text":"Cause","type":"text","marks":[{"type":"strong"}]},{"text":": Including too much conversation history","type":"text"}]},{"type":"paragraph","content":[{"text":"Solution","type":"text","marks":[{"type":"strong"}]},{"text":": Limit to last 2-3 relevant exchanges, summarize older context.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Advanced: Custom Routing Rules","type":"text"}]},{"type":"paragraph","content":[{"text":"To customize routing beyond default logic, edit ","type":"text"},{"text":"scripts/route_request.py","type":"text","marks":[{"type":"code_inline"}]},{"text":":","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"python"},"content":[{"text":"# Add custom patterns\nFORCE_CORTEX_PATTERNS = [\n \"snowflake\",\n \"cortex\",\n \"warehouse\",\n \"snowpark\"\n]\n\nFORCE_CLAUDE_PATTERNS = [\n \"local file\",\n \"git commit\",\n \"python script\" # unless Snowpark\n]","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"References","type":"text"}]},{"type":"paragraph","content":[{"text":"See ","type":"text"},{"text":"references/","type":"text","marks":[{"type":"code_inline"}]},{"text":" directory for:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"cortex-cli-reference.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" - Full Cortex CLI documentation","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"routing-examples.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" - More routing decision examples","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"session-file-format.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" - Cortex session file structure","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"troubleshooting-guide.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" - Extended troubleshooting","type":"text"}]}]}]}]},"metadata":{"date":"2026-11-02","name":"cortex-code","author":"@skillopedia","source":{"stars":63,"repo_name":"subagent-cortex-code","origin_url":"https://github.com/Snowflake-Labs/subagent-cortex-code/blob/HEAD/skills/cortex-code/SKILL.md","repo_owner":"Snowflake-Labs","body_sha256":"18bc74cfce1c4fedfb03c0289df8bd7dd2b3c9bc2848df457ba217aa1239e454","cluster_key":"79ecc36261709a33282be47b62345de63584bc9eb1c73bc3487af62b82b323b2","clean_bundle":{"format":"clean-skill-bundle-v1","source":"Snowflake-Labs/subagent-cortex-code/skills/cortex-code/SKILL.md","text_bytes":90231,"source_commit":"f3557faf20581ece1c46ab1a10b8041a4bb88532","skills_sh_rank":10060,"attachment_count":14,"text_attachments":14,"attachment_storage":"skill_attachments","binary_attachments":0},"cluster_size":1,"skill_md_path":"skills/cortex-code/SKILL.md","import_metadata":{"date":"2026-11-02","author":"@skillopedia","version":"v1","category":"data-analytics","category_label":"Data"},"exact_dupes_collapsed_into_this":0},"license":"Proprietary. See LICENSE for complete terms","version":"v1","category":"data-analytics","metadata":{"author":"Snowflake Integration Team","version":"1.0.0","compatibility":"Requires Cortex Code CLI installed and configured"},"import_tag":"skills-sh-top20k-2026-11","description":"Routes Snowflake-related operations to Cortex Code CLI for specialized Snowflake expertise. Use when user asks about Snowflake databases, data warehouses, SQL queries on Snowflake, Cortex AI features, Snowpark, dynamic tables, data governance in Snowflake, Snowflake security, or mentions \"Cortex\" explicitly. Do NOT use for general programming, local file operations, non-Snowflake databases, web development, or infrastructure tasks unrelated to Snowflake."}},"renderedAt":1793671375617}

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.