""" curl_exec tool — execute HTTP calls via subprocess, index responses, return truncated result. Parses curl command string, executes against live EC2 server, auto-injects session cookies, indexes full response into episode BM25 store, returns smart-truncated observation. """ from __future__ import annotations import json import re import shlex import subprocess import time from typing import Any from urllib.parse import urlparse # --------------------------------------------------------------------------- # Truncation constants # --------------------------------------------------------------------------- NONJSON_MAX_CHARS = 3000 # HTML / plain text truncation (raised for CSRF token visibility) ARRAY_PREVIEW_ITEMS = 2 # How many items to show in large arrays ARRAY_LARGE_THRESHOLD = 3 # Arrays >= this size are truncated # --------------------------------------------------------------------------- # Curl command parser # --------------------------------------------------------------------------- def parse_curl_command(command: str) -> dict: """ Parse a curl command string into components. Returns dict with keys: method, url, headers, body, data_type """ # Normalize: remove newline continuations command = re.sub(r"\\\s*\n\s*", " ", command) try: tokens = shlex.split(command) except ValueError: # Fall back to simple split if shlex fails tokens = command.split() if not tokens or tokens[0] != "curl": raise ValueError(f"Not a curl command: {command[:100]}") result: dict = { "method": "GET", "url": None, "headers": {}, "body": None, "data_type": None, # "json" | "form" | None } i = 1 while i < len(tokens): tok = tokens[i] if tok in ("-X", "--request") and i + 1 < len(tokens): result["method"] = tokens[i + 1].upper() i += 2 elif tok in ("-H", "--header") and i + 1 < len(tokens): header = tokens[i + 1] if ":" in header: name, _, value = header.partition(":") result["headers"][name.strip().lower()] = value.strip() i += 2 elif tok in ("-d", "--data", "--data-raw", "--data-binary") and i + 1 < len(tokens): result["body"] = tokens[i + 1] if result["method"] == "GET": result["method"] = "POST" i += 2 elif tok == "--data-urlencode" and i + 1 < len(tokens): # Append to existing body existing = result.get("body") or "" if existing: result["body"] = existing + "&" + tokens[i + 1] else: result["body"] = tokens[i + 1] if result["method"] == "GET": result["method"] = "POST" i += 2 elif tok in ("-F", "--form") and i + 1 < len(tokens): existing = result.get("body") or "" if existing: result["body"] = existing + "&" + tokens[i + 1] else: result["body"] = tokens[i + 1] if result["method"] == "GET": result["method"] = "POST" i += 2 elif tok in ("-u", "--user") and i + 1 < len(tokens): i += 2 # skip basic auth for now elif tok in ("-L", "--location", "-s", "--silent", "-v", "--verbose", "-k", "--insecure", "--compressed", "-g", "--globoff"): i += 1 elif tok in ("-o", "--output", "--max-time", "--connect-timeout", "--retry", "-A", "--user-agent", "-e", "--referer"): i += 2 # skip flag + value elif not tok.startswith("-") and result["url"] is None: result["url"] = tok.strip("'\"") i += 1 elif tok.startswith("http"): result["url"] = tok.strip("'\"") i += 1 else: i += 1 # Infer data_type from content-type header ct = result["headers"].get("content-type", "") if "application/json" in ct: result["data_type"] = "json" elif "application/x-www-form-urlencoded" in ct or "multipart/form-data" in ct: result["data_type"] = "form" elif result["body"]: # Guess from body if result["body"].strip().startswith("{") or result["body"].strip().startswith("["): result["data_type"] = "json" else: result["data_type"] = "form" return result # --------------------------------------------------------------------------- # Smart truncation # --------------------------------------------------------------------------- def smart_truncate(body_text: str, content_type: str = "") -> Any: """ Apply truncation rules to a response body string. Rules (first match wins): 1. Non-JSON → truncate to NONJSON_MAX_CHARS 2. JSON primitive (str/int/bool/null) → never truncate 3. Error (detected by content) → never truncate 4. JSON object/array with no large arrays → return as-is 5. JSON with large array → keep first ARRAY_PREVIEW_ITEMS, add _list_truncated note """ if not body_text: return "" # Rule 1: non-JSON if "application/json" not in content_type and not _looks_like_json(body_text): return body_text[:NONJSON_MAX_CHARS] # Try to parse as JSON try: parsed = json.loads(body_text) except (json.JSONDecodeError, ValueError): return body_text[:NONJSON_MAX_CHARS] # Rule 2: JSON primitive if not isinstance(parsed, (dict, list)): return parsed # Rule 3: detect error (4xx/5xx already handled by caller; this checks body content) if isinstance(parsed, dict) and ("message" in parsed or "error" in parsed): return parsed # never truncate errors # Rules 4 and 5 return _truncate_json(parsed) def _looks_like_json(text: str) -> bool: stripped = text.strip() return stripped.startswith("{") or stripped.startswith("[") or stripped.startswith('"') def _truncate_json(obj: Any) -> Any: if isinstance(obj, list): if len(obj) >= ARRAY_LARGE_THRESHOLD: return { "items": obj[:ARRAY_PREVIEW_ITEMS], "_list_truncated": { "shown": ARRAY_PREVIEW_ITEMS, "total": len(obj), "note": ( f"Showing {ARRAY_PREVIEW_ITEMS} of {len(obj)} items. " "Use search_episode_data() to find a specific item from this response." ), }, } return obj if isinstance(obj, dict): result = {} for k, v in obj.items(): if isinstance(v, list) and len(v) >= ARRAY_LARGE_THRESHOLD: result[k] = v[:ARRAY_PREVIEW_ITEMS] result["_list_truncated"] = { "field": k, "shown": ARRAY_PREVIEW_ITEMS, "total": len(v), "note": ( f"Showing {ARRAY_PREVIEW_ITEMS} of {len(v)} items. " "Use search_episode_data() to find a specific item from this response." ), } else: result[k] = v return result return obj # --------------------------------------------------------------------------- # Cookie injection # --------------------------------------------------------------------------- def _inject_cookies(headers: dict, session_state: dict) -> dict: """Inject cookies from session_state into the request headers.""" headers = dict(headers) # copy # Collect cookie values cookie_parts = [] for key, value in session_state.items(): if key.lower() in ("phpsessid", "sessid", "session", "cookie", "mage-cache-sessid", "private_content_version", "form_key", "PHPSESSID"): cookie_parts.append(f"{key}={value}") # Check if there's a raw cookie header already existing = headers.get("cookie", "") if cookie_parts: all_cookies = existing + ("; " if existing else "") + "; ".join(cookie_parts) headers["cookie"] = all_cookies return headers # --------------------------------------------------------------------------- # Session state extraction # --------------------------------------------------------------------------- def _extract_set_cookies(response_headers: dict, session_state: dict) -> None: """Extract Set-Cookie headers into session_state.""" for name, value in response_headers.items(): if name.lower() == "set-cookie": # Parse "NAME=VALUE; Path=...; ..." cookies = value.split(";") if cookies: kv = cookies[0].strip() if "=" in kv: k, _, v = kv.partition("=") session_state[k.strip()] = v.strip() def _extract_tokens_from_body(body: Any, session_state: dict) -> None: """Extract auth tokens from JSON response bodies into session_state.""" if isinstance(body, str) and len(body) > 10 and len(body) < 500: # Likely a token (Magento returns bare quoted strings for auth tokens) stripped = body.strip('"').strip() if re.match(r"^[A-Za-z0-9_\-\.]{20,}$", stripped): session_state["_last_token"] = stripped if isinstance(body, dict): for key in ("access_token", "token", "cart_id", "form_key"): if key in body and body[key]: session_state[key] = body[key] # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def curl_exec(command: str, session_state: dict, episode_store: dict, app_base_url: str = "") -> dict: """ Parse and execute a curl command against the live app. Args: command: Full curl command string session_state: Current session state (cookies/tokens), mutated in place episode_store: Per-episode store for BM25 indexing, mutated in place app_base_url: Base URL to validate requests against Returns: {status_code, headers, body} with smart-truncated body """ try: parsed = parse_curl_command(command) except Exception as e: return {"status_code": -1, "headers": {}, "body": f"curl parse error: {e}", "error": str(e)} if not parsed["url"]: return {"status_code": -1, "headers": {}, "body": "No URL in curl command", "error": "missing url"} # Inject session cookies parsed["headers"] = _inject_cookies(parsed["headers"], session_state) # Build actual curl args args = ["curl", "-s", "-i", "-L", "--max-time", "15"] args += ["-X", parsed["method"]] args += [parsed["url"]] for h_name, h_val in parsed["headers"].items(): args += ["-H", f"{h_name}: {h_val}"] if parsed["body"]: args += ["-d", parsed["body"]] try: result = subprocess.run( args, capture_output=True, text=True, timeout=20, ) raw_output = result.stdout except subprocess.TimeoutExpired: return {"status_code": -1, "headers": {}, "body": "Request timed out (20s)", "error": "timeout"} except Exception as e: return {"status_code": -1, "headers": {}, "body": f"subprocess error: {e}", "error": str(e)} # Parse HTTP response: headers + body split at blank line status_code = 0 resp_headers: dict[str, str] = {} body_text = "" if raw_output: # Find status line (handle redirects: multiple HTTP/ headers) lines = raw_output.split("\r\n") if "\r\n" in raw_output else raw_output.split("\n") header_lines = [] body_lines = [] in_body = False last_status = 0 for line in lines: if in_body: body_lines.append(line) elif line.startswith("HTTP/"): # Could be redirect status; keep last parts = line.split(" ", 2) if len(parts) >= 2: try: last_status = int(parts[1]) except ValueError: pass header_lines = [] # reset headers for this response elif line.strip() == "": if last_status: # we've seen at least one status line in_body = True else: header_lines.append(line) status_code = last_status body_text = "\n".join(body_lines).strip() for h_line in header_lines: if ":" in h_line: h_name, _, h_val = h_line.partition(":") resp_headers[h_name.strip().lower()] = h_val.strip() # Extract cookies / tokens into session_state _extract_set_cookies(resp_headers, session_state) # Try to parse body as JSON resp_ct = resp_headers.get("content-type", "") parsed_body: Any = body_text try: parsed_body = json.loads(body_text) if body_text else "" except (json.JSONDecodeError, ValueError): parsed_body = body_text # Distil HTML responses into structured compact form is_html_response = "text/html" in resp_ct if is_html_response and isinstance(parsed_body, str) and parsed_body: from .html_distiller import distill_html, distill_html_compact distilled = distill_html(parsed_body, base_url=parsed["url"]) # Auto-extract form_key from HTML forms into session_state for reuse for form in distilled.get("forms", []): fk = form.get("fields", {}).get("form_key") if fk and fk != "hidden": session_state["form_key"] = fk break # Store the full distilled dict (not raw HTML) for search_episode_data raw_body_for_store = distilled # What we return to the agent is the compact text summary truncated_body: Any = distill_html_compact(parsed_body, base_url=parsed["url"]) else: raw_body_for_store = parsed_body # Extract tokens from body (only for non-HTML responses) _extract_tokens_from_body(parsed_body, session_state) # Apply smart truncation if status_code >= 400: truncated_body = parsed_body else: body_for_truncation = body_text if isinstance(parsed_body, str) else json.dumps(parsed_body) truncated_body = smart_truncate(body_for_truncation, resp_ct) # Index into episode BM25 store _index_into_episode_store( episode_store=episode_store, request_body=parsed["body"], response_body=raw_body_for_store, url=parsed["url"], method=parsed["method"], status_code=status_code, ) return { "status_code": status_code, "headers": resp_headers, "body": truncated_body, # _judge_body: full structured body for the judge (not shown to the model) # For HTML: the distilled dict; for JSON/text: same as body "_judge_body": raw_body_for_store, } # --------------------------------------------------------------------------- # Episode store indexing # --------------------------------------------------------------------------- def _index_into_episode_store(episode_store: dict, request_body: Any, response_body: Any, url: str, method: str, status_code: int) -> None: """ Index request/response into the episode store for search_episode_data(). Three parallel structures are maintained: bm25_corpus — truncated text strings for BM25 / embedding (lean, fast) bm25_metadata — url/method/status_code per entry (no body, saves memory) episode_raw_bodies — {index: full_untruncated_response_body} for retrieval """ if "bm25_corpus" not in episode_store: episode_store["bm25_corpus"] = [] episode_store["bm25_metadata"] = [] episode_store["episode_raw_bodies"] = {} def _to_text(obj: Any) -> str: if obj is None: return "" if isinstance(obj, str): return obj return json.dumps(obj) # Lean text for BM25 / embedding — cap at 2000 chars so embeddings stay within # the model's token limit without losing the key signal (url + first part of body). # For distilled HTML (stored as a dict), serialize the distilled form — it's already # compact (text content, blob keys, form actions) rather than raw HTML. resp_text = _to_text(response_body) lean_resp = resp_text[:2000] if len(resp_text) > 2000 else resp_text entry_text = ( f"url: {url} method: {method} status: {status_code} " f"request: {_to_text(request_body)} response: {lean_resp}" ) idx = len(episode_store["bm25_corpus"]) episode_store["bm25_corpus"].append(entry_text) episode_store["bm25_metadata"].append({ "url": url, "method": method, "status_code": status_code, }) # Store full untruncated body keyed by index — never truncated episode_store["episode_raw_bodies"][idx] = response_body