"""Assistant module for Togen LLM chat (DVI-529). Provides a pluggable provider registry (ASSISTANT_PROVIDERS), settings persistence for global admin config and per-user connection overrides, an availability gate (is_assistant_available), a chat proxy, and a vision-based field-extraction helper used by the Claude-vision OCR backend. Architecture note (D1): ASSISTANT_PROVIDERS is intentionally open-ended so a future full-MCP provider can be registered here without touching routes or UI. The dispatch branch in chat() keys on provider id; MCP providers would add their own branch. """ from __future__ import annotations import json import os import re import stat import threading import uuid as _uuid from datetime import datetime, timezone from pathlib import Path from typing import Any # --------------------------------------------------------------------------- # Provider registry # --------------------------------------------------------------------------- ASSISTANT_PROVIDERS: dict[str, dict] = { "anthropic": { "id": "anthropic", "label": "Claude (Anthropic API)", "models": [ {"id": "claude-opus-4-7", "label": "Claude Opus 4.7"}, {"id": "claude-sonnet-4-6", "label": "Claude Sonnet 4.6"}, {"id": "claude-haiku-4-5-20251001", "label": "Claude Haiku 4.5"}, ], "default_model": "claude-sonnet-4-6", # Where the user creates an API key for this provider (shown as a link # on the User Settings -> Assistant pane once a provider is selected). "docs_url": "https://console.anthropic.com/settings/keys", }, } DEFAULT_PROVIDER = "anthropic" DEFAULT_MODEL = "claude-sonnet-4-6" # --------------------------------------------------------------------------- # Settings files # --------------------------------------------------------------------------- ASSISTANT_SETTINGS_FILE = Path( os.environ.get( "ASSISTANT_SETTINGS_FILE", Path(__file__).resolve().parent / "assistant_settings.json", ) ) DEFAULT_ASSISTANT_SETTINGS: dict[str, Any] = { "enabled": False, "accessible_to_users": False, "default_context_md": "", # org-level api_key is optional; absent means org key not configured } def load_assistant_settings() -> dict: """Return global assistant settings, falling back to defaults.""" settings = dict(DEFAULT_ASSISTANT_SETTINGS) if ASSISTANT_SETTINGS_FILE.is_file(): try: saved = json.loads(ASSISTANT_SETTINGS_FILE.read_text()) if isinstance(saved, dict): if isinstance(saved.get("enabled"), bool): settings["enabled"] = saved["enabled"] if isinstance(saved.get("accessible_to_users"), bool): settings["accessible_to_users"] = saved["accessible_to_users"] if isinstance(saved.get("default_context_md"), str): settings["default_context_md"] = saved["default_context_md"] if saved.get("api_key"): settings["api_key"] = saved["api_key"] except (json.JSONDecodeError, OSError): pass return settings def save_assistant_settings(settings: dict) -> None: """Persist global assistant settings. File is 0600 (key-bearing).""" ASSISTANT_SETTINGS_FILE.write_text(json.dumps(settings, indent=2)) ASSISTANT_SETTINGS_FILE.chmod(stat.S_IRUSR | stat.S_IWUSR) # --------------------------------------------------------------------------- # Per-user connection helpers # --------------------------------------------------------------------------- def get_user_assistant(user_settings: dict) -> dict: """Extract the assistant sub-dict from user settings (may be empty).""" return dict(user_settings.get("assistant") or {}) def set_user_assistant(user_settings: dict, assistant: dict) -> dict: """Return a copy of user_settings with the assistant sub-dict updated.""" updated = dict(user_settings) updated["assistant"] = assistant return updated # --------------------------------------------------------------------------- # Availability gate (plan §3.6) # --------------------------------------------------------------------------- def is_assistant_available(user_email: str, is_admin: bool, user_settings: dict) -> bool: """Return True if the assistant is available for this user. Admins: always available when the feature is enabled. Normal users: available when they have their own key, OR when the org key is configured and accessible_to_users is True. """ # Togen Admins always have the Assistant — the global enable/disable toggle # only gates normal users (per DVI-527 AC). Check the admin bypass before the # enabled gate so disabling globally never locks admins out. if is_admin: return True global_cfg = load_assistant_settings() if not global_cfg.get("enabled"): return False ua = get_user_assistant(user_settings) if ua.get("api_key"): return True if global_cfg.get("accessible_to_users") and global_cfg.get("api_key"): return True return False def is_assistant_connected(user_email: str, is_admin: bool, user_settings: dict) -> bool: """Return True only when the assistant is both entitled AND has a usable key. `is_assistant_available` is an *entitlement* gate — admins pass it unconditionally and org-key access passes it on the flag alone. That is not enough to actually run a request: without an effective key, chat fails with "No API key configured". This stricter check is what the topbar indicator (green vs red) and the chat gate use, so an entitled-but-unconfigured user (e.g. an admin who never set a key) sees red and the "configure your key" prompt instead of an input box that silently errors. """ if not is_assistant_available(user_email, is_admin, user_settings): return False return _effective_key(user_settings) is not None # --------------------------------------------------------------------------- # Effective key resolution # --------------------------------------------------------------------------- def _effective_key(user_settings: dict) -> str | None: """Return the API key to use for a request (user key beats org key).""" ua = get_user_assistant(user_settings) if ua.get("api_key"): return ua["api_key"] global_cfg = load_assistant_settings() return global_cfg.get("api_key") or None def _effective_context_md(user_settings: dict) -> str: """Return the context markdown to use (user override beats admin default).""" ua = get_user_assistant(user_settings) if ua.get("context_md") is not None: return ua["context_md"] global_cfg = load_assistant_settings() return global_cfg.get("default_context_md") or "" # --------------------------------------------------------------------------- # Chat proxy (non-streaming, D4) # --------------------------------------------------------------------------- def chat(messages: list[dict], system: str, *, user_settings: dict) -> str: """Send messages to the configured provider and return the reply text. messages: list of {role, content} dicts (OpenAI/Anthropic message format). system: system prompt string. user_settings: the caller's user-settings dict (for key resolution). Raises AssistantUnavailable if no key is configured. Raises AssistantError on API failure. """ key = _effective_key(user_settings) if not key: raise AssistantUnavailable("No API key configured.") try: import anthropic as _anthropic except ImportError as exc: raise AssistantUnavailable( "anthropic package is not installed." ) from exc client = _anthropic.Anthropic(api_key=key) ua = get_user_assistant(user_settings) model = ua.get("model") or DEFAULT_MODEL try: response = client.messages.create( model=model, max_tokens=4096, system=system or "", messages=messages, ) return response.content[0].text except _anthropic.APIError as exc: raise AssistantError(str(exc)) from exc # --------------------------------------------------------------------------- # Vision field extraction (OCR Claude-vision backend) # --------------------------------------------------------------------------- def extract_fields(image_bytes: bytes, fields: list[str]) -> dict[str, str]: """Use Claude vision to extract named fields from an image. image_bytes: raw PNG/JPEG image bytes. fields: list of field names to locate in the image (e.g. ["customer_name", "job_no"]). Returns a dict mapping field names to extracted string values (empty string when not found). Raises AssistantUnavailable or AssistantError on failure. """ import base64 key = _get_org_key() if not key: raise AssistantUnavailable("No org API key configured for Claude-vision OCR.") try: import anthropic as _anthropic except ImportError as exc: raise AssistantUnavailable("anthropic package is not installed.") from exc client = _anthropic.Anthropic(api_key=key) fields_list = "\n".join(f"- {f}" for f in fields) if fields == ["full_text"]: prompt = ( "You are an OCR assistant. Transcribe ALL text visible in this image, " "preserving line breaks and layout as closely as possible. " "Return ONLY a JSON object with a single key 'full_text' whose value " "is the full transcribed text. No other keys, no markdown, no commentary." ) else: prompt = ( "You are an OCR assistant. Extract the following fields from the image " "of an engineering drawing title block. Return ONLY a JSON object mapping " f"field names to their string values. Fields to extract:\n{fields_list}\n\n" "If a field is not present or not legible, use an empty string. " "Return only the JSON object, no other text." ) b64 = base64.standard_b64encode(image_bytes).decode() try: response = client.messages.create( model=DEFAULT_MODEL, max_tokens=1024, messages=[ { "role": "user", "content": [ { "type": "image", "source": { "type": "base64", "media_type": "image/png", "data": b64, }, }, {"type": "text", "text": prompt}, ], } ], ) raw = response.content[0].text.strip() # Strip markdown code fences if the model added them if raw.startswith("```"): raw = raw.split("```")[1] if raw.startswith("json"): raw = raw[4:] result = json.loads(raw) return {f: str(result.get(f, "")) for f in fields} except (json.JSONDecodeError, KeyError, IndexError) as exc: raise AssistantError(f"Unexpected response format from Claude vision: {exc}") from exc except _anthropic.APIError as exc: raise AssistantError(str(exc)) from exc def _get_org_key() -> str | None: """Return the org-level API key, or None.""" return load_assistant_settings().get("api_key") or None # --------------------------------------------------------------------------- # Vision image interpretation (DVI-867 Phase 1) # --------------------------------------------------------------------------- _INTERPRET_MAX_LONG_EDGE = 1568 # cap before sending to LLM (token budget) _INTERPRET_JPEG_QUALITY = 85 def _downscale_image(image_bytes: bytes) -> bytes: """Re-encode image to JPEG, capped at _INTERPRET_MAX_LONG_EDGE on the long edge. Accepts any PIL-readable format; returns JPEG bytes. Falls back to the original bytes if Pillow is unavailable or the image is already small enough. """ try: from io import BytesIO from PIL import Image img = Image.open(BytesIO(image_bytes)) w, h = img.size long_edge = max(w, h) if long_edge > _INTERPRET_MAX_LONG_EDGE: scale = _INTERPRET_MAX_LONG_EDGE / long_edge img = img.resize((int(w * scale), int(h * scale)), Image.LANCZOS) if img.mode not in ("RGB", "L"): img = img.convert("RGB") buf = BytesIO() img.save(buf, format="JPEG", quality=_INTERPRET_JPEG_QUALITY) return buf.getvalue() except Exception: return image_bytes def interpret_image( image_bytes: bytes, prompt: str, history: list[dict] | None = None, *, user_settings: dict, ) -> str: """Interpret a single image using the configured Vision assistant model. Delegates to interpret_images with a single-element list; behavior is unchanged. """ return interpret_images([image_bytes], prompt, history, user_settings=user_settings) def interpret_images( images: list[bytes], prompt: str, history: list[dict] | None = None, *, user_settings: dict, ) -> str: """Interpret an ordered sequence of images using the configured Vision assistant model. images: ordered list of raw image bytes (any PIL-readable format). Each image is preceded by a 'Frame k/N' text label so the model can reference frames. prompt: user question / instruction for this turn. history: prior conversation turns [{role, content}] — images are pinned on the first user turn as context for multi-turn sessions. user_settings: caller's settings dict (for key + model resolution). Raises AssistantUnavailable if no key is configured or provider not supported. Raises AssistantError on API failure. """ import base64 if not images: raise AssistantUnavailable("No images provided for interpretation.") # --- model resolution (mirrors interpret_image) --- raw_model = (user_settings.get("vision") or {}).get("assistant_model", "") if raw_model.startswith("system:"): model_id = raw_model[len("system:"):] key = _effective_key_system_only() elif raw_model.startswith("user:"): model_id = raw_model[len("user:"):] ua = get_user_assistant(user_settings) key = ua.get("api_key") or None else: model_id = DEFAULT_MODEL key = _effective_key(user_settings) if not model_id: model_id = DEFAULT_MODEL if not key: key = _effective_key(user_settings) if not key: raise AssistantUnavailable("No API key configured for Vision interpretation.") provider_id = "anthropic" for prov in ASSISTANT_PROVIDERS.values(): for m in prov.get("models", []): if m["id"] == model_id: provider_id = prov["id"] break if provider_id != "anthropic": raise AssistantUnavailable( f"Vision interpretation is not yet configured for provider '{provider_id}'." ) try: import anthropic as _anthropic except ImportError as exc: raise AssistantUnavailable("anthropic package is not installed.") from exc # --- build interleaved label + image blocks --- # Labels are added only for multi-frame sequences so that single-image # callers (interpret_image) see the same message structure as before. n = len(images) frame_content: list[dict] = [] for k, img_bytes in enumerate(images, start=1): scaled = _downscale_image(img_bytes) b64 = base64.standard_b64encode(scaled).decode() if n > 1: frame_content.append({"type": "text", "text": f"Frame {k}/{n}"}) frame_content.append({ "type": "image", "source": {"type": "base64", "media_type": "image/jpeg", "data": b64}, }) messages: list[dict] = [] if history: first_user_injected = False for turn in history: if turn.get("role") == "user" and not first_user_injected: messages.append({ "role": "user", "content": frame_content + [{"type": "text", "text": str(turn.get("content", ""))}], }) first_user_injected = True else: messages.append({"role": turn["role"], "content": str(turn.get("content", ""))}) messages.append({"role": "user", "content": prompt}) else: messages.append({ "role": "user", "content": frame_content + [{"type": "text", "text": prompt}], }) client = _anthropic.Anthropic(api_key=key) try: response = client.messages.create( model=model_id, max_tokens=2048, messages=messages, ) return response.content[0].text except _anthropic.APIError as exc: raise AssistantError(str(exc)) from exc def _effective_key_system_only() -> str | None: """Return the org-level key (for system: model selections).""" return load_assistant_settings().get("api_key") or None # --------------------------------------------------------------------------- # Tool dispatch infrastructure (Phase A, DVI-557) # --------------------------------------------------------------------------- # # Architecture: # - TIER0_TOOL_DEFS — registry of read-only Tier-0 tools; each entry carries # the Anthropic tool-use schema plus internal `required_roles` for authz. # - dispatch_tool() — authorize (deny-by-default) → execute → audit. # - chat_agentic() — Anthropic tool-use loop; admin-only in Phase A. # - Audit log — append-only JSONL outside model reach. # # Security invariants: # - Keys never enter model context. # - No tool may read assistant_settings.json / user settings / env. # - Tool results are scrubbed: no filesystem paths, no secrets. # - Every call is audited regardless of authz outcome. # Paths mirroring the same env vars used in app.py — read once at import time. _PRODRUN_DIR = Path(os.environ.get("PRODRUN_DIR", "/var/www/html/prodrunmerge")) _SHARED_DIR = Path(os.environ.get("QR_SHARED_DIR", "/var/www/html/shared")) _WOR_DIR = Path(os.environ.get("WOR_DIR", "/var/www/html/wor")) _APP_DIR = Path(__file__).resolve().parent TOOL_AUDIT_LOG = Path( os.environ.get("TOOL_AUDIT_LOG", str(_APP_DIR / "tool_audit.jsonl")) ) _SAFE_TOOL_ARG_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9 _\-\.]{0,79}$") _audit_lock = threading.Lock() def _audit_append(entry: dict) -> None: """Append one JSON line to the audit log; swallows exceptions.""" try: with _audit_lock: with TOOL_AUDIT_LOG.open("a", encoding="utf-8") as fh: fh.write(json.dumps(entry) + "\n") except Exception: pass # Tier-0 tool definitions. `required_roles`: empty = any authenticated user; # non-empty = user must hold at least one of the listed session role flags. _TIER0_TOOL_DEFS: dict[str, dict] = { "prodrun_status": { "required_roles": ["prod_run", "togen_admin"], "anthropic_def": { "name": "prodrun_status", "description": ( "Get the current status of a production run merge job. " "Returns status, message, and result summary." ), "input_schema": { "type": "object", "properties": { "job_id": { "type": "string", "description": ( "Production run job ID — alphanumeric, spaces, " "dashes, or dots; max 80 characters." ), } }, "required": ["job_id"], }, }, }, "reprint_list_projects": { "required_roles": [], "anthropic_def": { "name": "reprint_list_projects", "description": "List available project folders for reprint or lookup, newest first.", "input_schema": { "type": "object", "properties": {}, "required": [], }, }, }, "wor_history": { "required_roles": [], "anthropic_def": { "name": "wor_history", "description": ( "List new and in-progress Work Order Request PDFs. " "Defaults to 'maintenance' type." ), "input_schema": { "type": "object", "properties": { "type": { "type": "string", "description": "WOR type.", "enum": ["maintenance", "mobilemachine", "safety", "preventive"], } }, "required": [], }, }, }, "manuals_lookup": { "required_roles": [], "anthropic_def": { "name": "manuals_lookup", "description": ( "Look up a manual previously associated with an Equipment, " "Asset, or System name." ), "input_schema": { "type": "object", "properties": { "equipment": { "type": "string", "description": "Equipment, asset, or system name.", } }, "required": ["equipment"], }, }, }, } # Anthropic-format tool list for use in client.messages.create(tools=...) TIER0_ANTHROPIC_TOOLS: list[dict] = [ d["anthropic_def"] for d in _TIER0_TOOL_DEFS.values() ] # --------------------------------------------------------------------------- # Tier-1 (mutating) tool registry (Phase B, DVI-560) # --------------------------------------------------------------------------- # These tools write data. They are only surfaced to togen_admin users (Phase B # is admin-only) and require explicit per-action human confirmation before # execution. Tier-2 routes (togen-admin writes, CSV-schema paths) are not # registered here and cannot be reached through the agentic path. _TIER1_TOOL_DEFS: dict[str, dict] = { "qr_submit": { "required_roles": ["takeoff_workflow", "togen_admin"], "tier": 1, "display_name": "Submit QR Merge Job", "make_display_args": lambda a: {"Folder": a.get("folder", "")}, "anthropic_def": { "name": "qr_submit", "description": ( "Submit a QR code merge job for a project folder. " "Merges shop drawings with carrier sheets and writes results to the shared folder. " "Requires explicit user confirmation before execution." ), "input_schema": { "type": "object", "properties": { "folder": { "type": "string", "description": "Project folder name (alphanumeric, spaces, dashes, dots; max 80 chars).", } }, "required": ["folder"], }, }, }, "duplicate_folder": { "required_roles": ["takeoff_workflow", "togen_admin"], "tier": 1, "display_name": "Duplicate Project Folder", "make_display_args": lambda a: { "Source": a.get("folder", ""), "New name": a.get("new_folder", ""), }, "anthropic_def": { "name": "duplicate_folder", "description": ( "Duplicate a project folder under a new name, copying all documents, CSV, and metadata. " "Requires explicit user confirmation before execution." ), "input_schema": { "type": "object", "properties": { "folder": { "type": "string", "description": "Source project folder name.", }, "new_folder": { "type": "string", "description": "New folder name (must not already exist).", }, }, "required": ["folder", "new_folder"], }, }, }, "prodrun_submit": { "required_roles": ["prod_run", "togen_admin"], "tier": 1, "display_name": "Submit Production Run", "make_display_args": lambda a: {"Folder": a.get("folder", "")}, "anthropic_def": { "name": "prodrun_submit", "description": ( "Submit a Production Run merge job. Starts the merge worker for the given folder. " "Requires explicit user confirmation before execution." ), "input_schema": { "type": "object", "properties": { "folder": { "type": "string", "description": "Production run job folder name.", } }, "required": ["folder"], }, }, }, "scs_save_to_project": { "required_roles": ["takeoff_workflow", "togen_admin"], "tier": 1, "display_name": "Save Carrier Sheet to Project", "make_display_args": lambda a: { "Job": a.get("job_id", ""), "Project folder": a.get("folder_name", ""), "Product code": a.get("product_code", ""), }, "anthropic_def": { "name": "scs_save_to_project", "description": ( "Save a Single Carrier Sheet (from an existing creation session) to a project folder, " "updating the project CSV with the carrier sheet URL. " "Requires explicit user confirmation before execution." ), "input_schema": { "type": "object", "properties": { "job_id": { "type": "string", "description": "SCS job ID from a completed carrier creation session.", }, "folder_name": { "type": "string", "description": "Target project folder name in the shared directory.", }, "product_code": { "type": "string", "description": "Product code to match or append in the project CSV.", }, "append_row": { "type": "boolean", "description": "Append a new CSV row if the product code is not found (default: false).", }, }, "required": ["job_id", "folder_name", "product_code"], }, }, }, "create_carrier": { "required_roles": [], # login_required — any authenticated user; admin-gate is at the agentic layer "tier": 1, "display_name": "Create Carrier Sheet", "make_display_args": lambda a: { "Job number": a.get("job_number", a.get("job_id", "")), "Product code": a.get("product_code", ""), "Customer": a.get("customer_name", ""), }, "anthropic_def": { "name": "create_carrier", "description": ( "Generate a QR carrier sheet for an existing upload session (job_id). " "The shop drawing PDF must already have been uploaded via the Create Carrier UI first. " "Optionally overrides carrier field values. " "Requires explicit user confirmation before execution." ), "input_schema": { "type": "object", "properties": { "job_id": { "type": "string", "description": "Job ID from a completed file-upload session (returned by the UI after uploading).", }, "job_number": {"type": "string", "description": "Job number override."}, "product_code": {"type": "string", "description": "Product code override."}, "customer_name": {"type": "string", "description": "Customer name override."}, "project_name": {"type": "string", "description": "Project name override."}, }, "required": ["job_id"], }, }, }, } # Merged registry: Tier-0 (read-only) + Tier-1 (mutating). Used for authz. _ALL_TOOL_DEFS: dict[str, dict] = {**_TIER0_TOOL_DEFS, **_TIER1_TOOL_DEFS} # Callbacks registered by app.py at startup — one per Tier-1 tool. # Signature: fn(args: dict, user: dict) -> dict _tier1_callbacks: dict[str, Any] = {} def register_tier1_callback(tool_id: str, fn: Any) -> None: """Register the executor for a Tier-1 tool (called by app.py at startup).""" _tier1_callbacks[tool_id] = fn def _authorize_tool(tool_id: str, user: dict) -> bool: """Return True when user holds at least one required role for this tool. Deny-by-default: unknown tool_ids return False. Empty required_roles list means any authenticated user is permitted (mirrors @login_required). Mirrors the same role flags as the underlying Flask route decorators. """ defn = _ALL_TOOL_DEFS.get(tool_id) if defn is None: return False required = defn["required_roles"] if not required: return True return any(user.get(r) for r in required) def has_any_tool_access(user: dict) -> bool: """Return True if the user is authorized to call at least one registered tool. Used by the chat route (Phase C) to decide whether a non-admin user gets agentic mode — if they have access to at least one tool, they use chat_agentic(); otherwise they fall back to plain chat(). """ return any(_authorize_tool(tid, user) for tid in _ALL_TOOL_DEFS) # --------------------------------------------------------------------------- # Tool handler functions — read-only, no mutations # --------------------------------------------------------------------------- def _tool_prodrun_status(args: dict, _user: dict) -> dict: """Return production run job status (safe summary only — no file paths).""" job_id = str(args.get("job_id", "")).strip() if not job_id or not _SAFE_TOOL_ARG_RE.match(job_id) or ".." in job_id: return {"error": "Invalid job_id."} status_path = _PRODRUN_DIR / job_id / "status.json" if not status_path.exists(): return {"status": "pending", "message": "Initializing…"} try: data = json.loads(status_path.read_text()) # Scrub: return only safe summary fields — no file system paths safe: dict[str, Any] = {} for k in ("status", "message", "drawing_merge", "label_merge"): if k in data: safe[k] = data[k] return safe except Exception: return {"error": "Could not read status."} def _tool_reprint_list_projects(_args: dict, _user: dict) -> dict: """Return project folder names from the shared directory.""" folders: list[str] = [] if _SHARED_DIR.is_dir(): try: for entry in sorted( _SHARED_DIR.iterdir(), key=lambda e: e.stat().st_mtime, reverse=True, ): if entry.is_dir() and not entry.name.startswith("."): folders.append(entry.name) except OSError: pass return {"folders": folders} def _tool_wor_history(args: dict, _user: dict) -> dict: """Return new and in-progress WOR PDF names for the requested type.""" wor_type = str(args.get("type", "maintenance")).strip() if wor_type not in {"maintenance", "mobilemachine", "safety", "preventive"}: wor_type = "maintenance" wor_base = _WOR_DIR / wor_type dirs: list[tuple[Path, str]] = [ (wor_base, "new"), (wor_base / "in_progress", "in_progress"), ] files: list[dict] = [] for directory, status in dirs: if not directory.is_dir(): continue for f in directory.glob("*.pdf"): try: mtime = datetime.utcfromtimestamp(f.stat().st_mtime) files.append({ "name": f.name, "date": mtime.strftime("%Y-%m-%dT%H:%M:%SZ"), "status": status, }) except OSError: continue files.sort(key=lambda x: x["date"], reverse=True) return {"files": files} def _tool_manuals_lookup(args: dict, _user: dict) -> dict: """Look up a manual by equipment name; returns name + association date only.""" equipment = str(args.get("equipment", "")).strip() if not equipment: return {"found": False} key = " ".join(equipment.lower().split()) index_file = _APP_DIR / "pm_manuals_index.json" if not index_file.is_file(): return {"found": False} try: data = json.loads(index_file.read_text()) except (json.JSONDecodeError, OSError): return {"found": False} entry = data.get(key) if not entry or not entry.get("filename"): return {"found": False} # Return safe fields only — no server-side paths return { "found": True, "equipment": entry.get("equipment", ""), "original_name": entry.get("original_name", entry["filename"]), "updated_at": entry.get("updated_at", ""), } _TOOL_HANDLERS: dict[str, Any] = { "prodrun_status": _tool_prodrun_status, "reprint_list_projects": _tool_reprint_list_projects, "wor_history": _tool_wor_history, "manuals_lookup": _tool_manuals_lookup, } def dispatch_tool(tool_id: str, args: dict, user: dict) -> dict: """Authorize, execute, and audit one tool call. Returns a result dict suitable for a tool_result message block. Never raises — errors are returned as {"error": "..."} dicts and audited. """ ts = datetime.now(timezone.utc).isoformat() email = (user or {}).get("email", "unknown") authz = _authorize_tool(tool_id, user) audit: dict[str, Any] = { "ts": ts, "user": email, "tool": tool_id, "args": args, "authz": authz, } if not authz: audit["outcome"] = "denied" _audit_append(audit) return {"error": f"Access denied: insufficient permissions for tool '{tool_id}'."} handler = _TOOL_HANDLERS.get(tool_id) if handler is None: audit["outcome"] = "unknown_tool" _audit_append(audit) return {"error": f"Unknown tool: '{tool_id}'."} try: result = handler(args, user) # Distinguish validated-input rejections from genuine successes so # audit reviewers can tell "blocked path traversal" from "returned data". audit["outcome"] = "input_error" if "error" in result else "success" _audit_append(audit) return result except Exception as exc: audit["outcome"] = f"handler_error: {exc}" _audit_append(audit) return {"error": "Tool execution failed."} # --------------------------------------------------------------------------- # Pending-action store (Phase B, DVI-560) # --------------------------------------------------------------------------- # When the agentic loop hits a Tier-1 tool call it stores conversation state # here, raises PendingActionRequired, and waits for user confirmation. Each # token is bound to the requesting user's email and expires after 10 minutes. _pending_actions: dict[str, dict] = {} _pending_lock = threading.Lock() _PENDING_ACTION_TTL = 600 # 10 minutes def _pending_purge_expired() -> None: """Remove expired tokens (caller must hold _pending_lock).""" now = datetime.now(timezone.utc).timestamp() expired = [k for k, v in _pending_actions.items() if now - v["ts"] > _PENDING_ACTION_TTL] for k in expired: del _pending_actions[k] def _pending_create( tool: str, args: dict, user_email: str, working_messages: list, tool_use_id: str, tier0_results: list, system: str, model: str, authorized_tools: list, ) -> str: """Store conversation state keyed by a fresh UUID token; return the token.""" token = str(_uuid.uuid4()) with _pending_lock: _pending_purge_expired() _pending_actions[token] = { "ts": datetime.now(timezone.utc).timestamp(), "tool": tool, "args": args, "user_email": user_email, "working_messages": working_messages, "tool_use_id": tool_use_id, "tier0_results": tier0_results, "system": system, "model": model, "authorized_tools": authorized_tools, } return token def _pending_consume(token: str, user_email: str) -> dict | None: """Pop and return a pending action; None if missing, expired, or wrong user. Leaves the entry in place when the email doesn't match so the correct user can still consume it. """ with _pending_lock: _pending_purge_expired() entry = _pending_actions.get(token) if entry is None: return None if entry.get("user_email") != user_email: return None # wrong user — leave token intact del _pending_actions[token] return entry # --------------------------------------------------------------------------- # Agentic chat — Anthropic tool-use loop (Phase A + B) # --------------------------------------------------------------------------- _AGENTIC_MAX_ROUNDS = 10 # safety cap: maximum tool-call rounds per request def _content_block_to_dict(block: Any) -> dict: """Convert an Anthropic SDK content block to a plain dict.""" if block.type == "text": return {"type": "text", "text": block.text} if block.type == "tool_use": return { "type": "tool_use", "id": block.id, "name": block.name, "input": dict(block.input), } # Fallback: try model_dump() for future block types, else drop try: return block.model_dump() except Exception: return {"type": block.type} def _agentic_loop( client: Any, working_messages: list, system: str, model: str, authorized_tools: list, user: dict, ) -> str: """Run the agentic tool-calling loop until end_turn or the round cap. Executes Tier-0 tools inline. Raises PendingActionRequired when the model calls a Tier-1 (mutating) tool so the caller can request user confirmation. Raises AssistantError on provider API failure. """ try: import anthropic as _anthropic except ImportError as exc: raise AssistantUnavailable("anthropic package is not installed.") from exc user_email = (user or {}).get("email", "") for _round in range(_AGENTIC_MAX_ROUNDS): try: response = client.messages.create( model=model, max_tokens=4096, system=system or "", messages=working_messages, tools=authorized_tools, ) except _anthropic.APIError as exc: raise AssistantError(str(exc)) from exc if response.stop_reason == "end_turn": final_text = "" for block in response.content: if block.type == "text": final_text += block.text return final_text if response.stop_reason == "tool_use": working_messages.append({ "role": "assistant", "content": [_content_block_to_dict(b) for b in response.content], }) # Collect partial text and separate tool calls by tier partial_text = "" tier0_calls: list = [] tier1_calls: list = [] for block in response.content: if block.type == "text": partial_text += block.text elif block.type == "tool_use": if block.name in _TIER1_TOOL_DEFS: tier1_calls.append(block) else: tier0_calls.append(block) # Execute Tier-0 (read-only) tools immediately tier0_results = [] for block in tier0_calls: result = dispatch_tool(block.name, dict(block.input), user) tier0_results.append({ "type": "tool_result", "tool_use_id": block.id, "content": json.dumps(result), }) if tier1_calls: # Pause on the first Tier-1 tool; store state for user confirmation t1 = tier1_calls[0] defn = _TIER1_TOOL_DEFS[t1.name] display_args = defn["make_display_args"](dict(t1.input)) token = _pending_create( tool=t1.name, args=dict(t1.input), user_email=user_email, working_messages=working_messages, tool_use_id=t1.id, tier0_results=tier0_results, system=system, model=model, authorized_tools=authorized_tools, ) raise PendingActionRequired( token=token, tool=t1.name, display_name=defn["display_name"], display_args=display_args, partial_reply=partial_text, ) working_messages.append({"role": "user", "content": tier0_results}) continue # max_tokens or any other stop_reason final_text = "" for block in response.content: if block.type == "text": final_text += block.text return final_text return "\n\n[Response truncated: tool-call limit reached.]" def chat_agentic( messages: list[dict], system: str, *, user: dict, user_settings: dict, ) -> str: """Agentic chat: Anthropic tool-use loop with per-tool authz and audit. Tier-0 (read-only) tools execute inline. Tier-1 (mutating) tools pause the loop and raise PendingActionRequired; call resume_agentic() after the user decides. user: session["user"] dict — used for per-tool authorization. user_settings: caller's user-settings dict — used for key resolution. Raises AssistantUnavailable / AssistantError on API failure. Raises PendingActionRequired when a mutating tool awaits user confirmation. """ key = _effective_key(user_settings) if not key: raise AssistantUnavailable("No API key configured.") try: import anthropic as _anthropic except ImportError as exc: raise AssistantUnavailable("anthropic package is not installed.") from exc client = _anthropic.Anthropic(api_key=key) ua = get_user_assistant(user_settings) model = ua.get("model") or DEFAULT_MODEL # Build authorized tool list (Tier-0 + Tier-1). dispatch_tool() still # enforces authz per-call; this prevents the model from requesting tools # the user can't use. authorized_tools = [ defn["anthropic_def"] for tool_id, defn in _ALL_TOOL_DEFS.items() if _authorize_tool(tool_id, user) ] # Prepend confirmation-contract preamble when Tier-1 tools are available tier1_names = [tid for tid in _TIER1_TOOL_DEFS if _authorize_tool(tid, user)] if tier1_names: names_str = ", ".join(tier1_names) preamble = ( "IMPORTANT — mutating tools: when you call any of the following tools " f"({names_str}), the user will see a confirmation prompt before the " "action executes. If they decline you will receive " '{"cancelled": true, "message": "Action cancelled by user."} as the ' "tool result; acknowledge this gracefully. " "Propose at most one mutating action per response turn." ) system = (preamble + "\n\n" + system).strip() if system else preamble return _agentic_loop( client, list(messages), system, model, authorized_tools, user ) def resume_agentic( token: str, confirmed: bool, user: dict, user_settings: dict, ) -> str: """Resume the agentic loop after the user confirms or cancels a Tier-1 action. Retrieves the pending conversation state, executes (or skips) the Tier-1 tool, and continues the loop until the model emits a final reply. Raises AssistantUnavailable / AssistantError on API failure. Raises PendingActionRequired if the resumed loop hits another Tier-1 tool. """ user_email = (user or {}).get("email", "") entry = _pending_consume(token, user_email) if entry is None: return "This action has expired or was already handled. Please start a new request." tool = entry["tool"] args = entry["args"] working_msgs = entry["working_messages"] tool_use_id = entry["tool_use_id"] tier0_results = entry["tier0_results"] system = entry["system"] model = entry["model"] authorized_tools = entry["authorized_tools"] ts = datetime.now(timezone.utc).isoformat() audit: dict[str, Any] = { "ts": ts, "user": user_email, "tool": tool, "args": args, "authz": True, } if not confirmed: audit["outcome"] = "cancelled_by_user" _audit_append(audit) tier1_result: dict = {"cancelled": True, "message": "Action cancelled by user."} else: cb = _tier1_callbacks.get(tool) if cb is None: audit["outcome"] = "no_callback" _audit_append(audit) tier1_result = {"error": f"No handler registered for '{tool}'. Contact your administrator."} else: try: tier1_result = cb(args, user) audit["outcome"] = "input_error" if "error" in tier1_result else "success" _audit_append(audit) except Exception as exc: audit["outcome"] = f"handler_error: {exc}" _audit_append(audit) tier1_result = {"error": "Tool execution failed."} # Combine tier-0 results (dispatched before the pause) + tier-1 result all_tool_results = list(tier0_results) + [{ "type": "tool_result", "tool_use_id": tool_use_id, "content": json.dumps(tier1_result), }] resumed_messages = list(working_msgs) + [ {"role": "user", "content": all_tool_results} ] key = _effective_key(user_settings) if not key: raise AssistantUnavailable("No API key configured.") try: import anthropic as _anthropic except ImportError as exc: raise AssistantUnavailable("anthropic package is not installed.") from exc client = _anthropic.Anthropic(api_key=key) return _agentic_loop(client, resumed_messages, system, model, authorized_tools, user) # --------------------------------------------------------------------------- # Key masking helper # --------------------------------------------------------------------------- def mask_key(key: str | None) -> dict: """Return a safe public representation of a key (never the raw value).""" if not key: return {"configured": False, "masked": ""} last4 = key[-4:] if len(key) >= 4 else "****" return {"configured": True, "masked": f"••••{last4}"} # --------------------------------------------------------------------------- # Exceptions # --------------------------------------------------------------------------- class AssistantUnavailable(RuntimeError): """Raised when the assistant cannot be used (not configured / gated out).""" class AssistantError(RuntimeError): """Raised on API or provider errors during a chat/vision call.""" class PendingActionRequired(Exception): """Raised by the agentic loop when a Tier-1 tool awaits user confirmation. Attributes: token: UUID string identifying the stored conversation state. tool: Tier-1 tool name. display_name: Human-readable action label for the confirmation card. display_args: Key-value pairs to render in the confirmation card. partial_reply: Any model text emitted before the tool call (may be ""). """ __slots__ = ("token", "tool", "display_name", "display_args", "partial_reply") def __init__( self, token: str, tool: str, display_name: str, display_args: dict, partial_reply: str = "", ) -> None: self.token = token self.tool = tool self.display_name = display_name self.display_args = display_args self.partial_reply = partial_reply