diff --git a/CHANGELOG.md b/CHANGELOG.md index 244853bd..195e1d63 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,14 @@ and this project adheres to [PEP 440](https://peps.python.org/pep-0440/) / ## [Unreleased] +### Fixed + +- **Artifact handler notify webhook 403 on Lambda.** Notify API keys are minted with `/api/artifact_handler_action/notify/*` like Drive/WhatsApp/PageIndex, so jvspatial webhook auth no longer 403s when the request path does not byte-match an exact agent id. The handler still hmac-binds the key id to this action. Exact-path keys remint on the next vault submit. + +- **Artifact handler notify 404 on Lambda.** `POST /api/artifact_handler_action/notify/{agent_id}` is imported with first-party routes before `get_app()`, and remounted if the live FastAPI app was already built, so jvforge's import callback is no longer a registry-only 404. + +- **Artifact handler notify awaits `create_task` return and time-bounds ready-answer generate.** After inline import, generate then send run as two sequential `jvspatial.create_task` calls; a non-None return is awaited so Lambda cannot 200-and-freeze the ready message. PageIndex desc lookup (5s) and ready-answer generate (12s) time out to a canned message, then the channel send still runs. + ### Added - **Harness excellence runtime (HP-02 … HP-12).** `jvagent.harness.runtime` is the store-backed source of truth for NativeCaller admission, snapshot-keyed caches, TurnRun journals, invocation ledger, durable outbox, session leases, host providers, skill manifests/isolation, traces, and the HP-12 deployment matrix. Process-local bus/caches remain fan-out; JSON/SQLite active-active is unsupported. Docs: `docs/HARNESS_DEPLOYMENT.md`, `docs/skill-isolation.md`. TurnRun checkpoints persist on `Interaction.observability_metrics`; loop resume skips completed IDEMPOTENT invocations; Claude skill staging is snapshot/digest-keyed and refuses untrusted isolation; mutating send/delete/bash tools declare `NON_RETRYABLE`; embed cancel marks TurnRun recovery. HostCapabilityProvider.invoke dispatches a registered host runner; IsolatedExecutor wraps approved backends with no subprocess fallback; dump_store/load_store persist the harness store; file/redis/dynamo lease adapters require an explicit client; skill signatures use HMAC compare_digest; CUCS harness evals live under `tests/conformance/cucs/`; CI adds conformance/two-worker/isolation/load lanes. @@ -33,6 +41,14 @@ and this project adheres to [PEP 440](https://peps.python.org/pep-0440/) / ### Changed +- **Artifact handler notify logs.** Dropped breadcrumb `logger.warning` traces (entered, remint, import start, send ok, canned fallback, `create_task` scheduled). Failures stay as `logger.error` / `logger.exception`. + +- **PageIndex LLM webhook path.** `POST /api/pageindex/interact/webhook/{agent_id}` replaces `/api/pageindex_retrieval_interact_action/interact/webhook/{agent_id}`. Keys remint on the next `get_webhook_url`. Old URLs 404; ship with matching jvforge. + +- **Artifact handler notify awaits channel send.** WhatsApp/Messenger ready messages are sent in the notify request (not `asyncio.create_task`) so Lambda does not freeze the send when the Function URL returns. A failed send leaves the job in the reverse index and returns **503** so jvforge retries. + +- **Artifact handler notify + ingest status.** `register_job` now fails the ingest when the reverse-index cannot be saved (instead of swallowing `save()` and leaving chats stuck on `queued`). The notify webhook reloads the action from the DB before `lookup_job`, logs unknown jobs, and returns **503** so jvforge retries. `check_ingest_status` still consults PageIndex first, then polls jvforge and pull-imports `webhook_failed` / `completed` artifacts; jvforge `failed` marks the vault job failed. Pending and vault entry statuses stay in sync via `apply_ingest_job_status`. + - **ResponseBus now enforces the single-egress latch.** The first delivered non-transient user stream chunk marks its `Interaction` as emitted, active chunks may finish that same stream, and any later independent user publish diff --git a/jvagent/action/artifact_handler_interact_action/artifact_handler_interact_action.py b/jvagent/action/artifact_handler_interact_action/artifact_handler_interact_action.py index 3328b126..50fd6c3e 100644 --- a/jvagent/action/artifact_handler_interact_action/artifact_handler_interact_action.py +++ b/jvagent/action/artifact_handler_interact_action/artifact_handler_interact_action.py @@ -36,6 +36,7 @@ import importlib import json +import logging import os import re import sys @@ -60,6 +61,8 @@ if False: from jvagent.action.interact.interact_walker import InteractWalker +logger = logging.getLogger(__name__) + def _register_orchestrator_vocabulary() -> None: """Declare vault tool results as a trusted directive source. @@ -561,6 +564,11 @@ async def execute(self, visitor: "InteractWalker") -> None: try: notification_url = await self.get_notify_webhook_url() except Exception: + logger.exception( + "artifact_handler notify: webhook url mint failed " + "agent_id=%s", + getattr(self, "agent_id", None), + ) notification_url = None if not notification_url: await visitor.add_directive( @@ -669,6 +677,10 @@ async def execute(self, visitor: "InteractWalker") -> None: filename=display_filename, ) except Exception: + logger.exception( + "artifact_handler execute: submit_ingest failed filename=%s", + filename, + ) failed.append(filename) continue @@ -955,10 +967,10 @@ async def get_notify_webhook_url( from jvagent.core.public_url import get_public_base_url from .webhook_auth import ( + ALLOWED_WEBHOOK_ENDPOINT_GLOB, ARTIFACT_HANDLER_NOTIFY_ROUTE_PREFIX, WEBHOOK_PERMISSION, get_or_create_system_user, - notify_endpoint_for_agent, ) base_url = (get_public_base_url() or "").strip().rstrip("/") @@ -974,7 +986,6 @@ async def get_notify_webhook_url( expected_url_base = ( f"{base_url}/api/{ARTIFACT_HANDLER_NOTIFY_ROUTE_PREFIX}/{agent_id}" ) - allowed_endpoint = notify_endpoint_for_agent(agent_id) def _key_scoped_to_agent(existing_key: Any) -> bool: if existing_key is None or not getattr( @@ -984,14 +995,7 @@ def _key_scoped_to_agent(existing_key: Any) -> bool: existing_eps = list( getattr(existing_key, "allowed_endpoints", None) or [] ) - if allowed_endpoint not in existing_eps: - return False - for ep in existing_eps: - if ARTIFACT_HANDLER_NOTIFY_ROUTE_PREFIX not in ep: - continue - if ep.endswith("*"): - return False - return True + return ALLOWED_WEBHOOK_ENDPOINT_GLOB in existing_eps prime_ctx = GraphContext(database=get_prime_database()) api_key_service = APIKeyService(context=prime_ctx) @@ -1037,7 +1041,7 @@ def _key_scoped_to_agent(existing_key: Any) -> bool: permissions=[WEBHOOK_PERMISSION], expires_in_days=None, allowed_ips=[allowed_ip] if allowed_ip else [], - allowed_endpoints=[allowed_endpoint], + allowed_endpoints=[ALLOWED_WEBHOOK_ENDPOINT_GLOB], key_prefix="jv_", ) @@ -1093,10 +1097,9 @@ async def register_job( "file_url": saved_url or "", } self.jvforge_job_index = index - try: - await self.save() - except Exception: - pass + await self._persist_job_index( + job_id=job_id, op="register_job", raise_on_error=True + ) async def lookup_job(self, job_id: str) -> Optional[Dict[str, Any]]: if not job_id: @@ -1105,6 +1108,25 @@ async def lookup_job(self, job_id: str) -> Optional[Dict[str, Any]]: entry = index.get(job_id) return dict(entry) if isinstance(entry, dict) else None + async def _persist_job_index( + self, *, job_id: str, op: str, raise_on_error: bool = False + ) -> None: + index = self.jvforge_job_index or {} + size = len(index) if isinstance(index, dict) else 0 + try: + await self.save() + except Exception: + logger.exception( + "artifact_handler %s save failed job_id=%s agent_id=%s index_size=%s", + op, + job_id, + getattr(self, "agent_id", None), + size, + ) + if raise_on_error: + raise + return + async def clear_job(self, job_id: str) -> None: if not job_id: return @@ -1112,10 +1134,7 @@ async def clear_job(self, job_id: str) -> None: if job_id in index: del index[job_id] self.jvforge_job_index = index - try: - await self.save() - except Exception: - pass + await self._persist_job_index(job_id=job_id, op="clear_job") async def mark_notified(self, job_id: str) -> None: if not job_id: @@ -1128,10 +1147,7 @@ async def mark_notified(self, job_id: str) -> None: entry["notified_at"] = _utc_iso() index[job_id] = entry self.jvforge_job_index = index - try: - await self.save() - except Exception: - pass + await self._persist_job_index(job_id=job_id, op="mark_notified") async def mark_notifying(self, job_id: str) -> None: if not job_id: @@ -1143,10 +1159,7 @@ async def mark_notifying(self, job_id: str) -> None: entry["notifying_at"] = _now_ts() index[job_id] = entry self.jvforge_job_index = index - try: - await self.save() - except Exception: - pass + await self._persist_job_index(job_id=job_id, op="mark_notifying") # ── Async jvforge ingest submission ── @@ -1199,6 +1212,11 @@ async def submit_ingest( try: notify = (await self.get_notify_webhook_url() or "").strip() except Exception: + logger.exception( + "artifact_handler notify: submit_ingest webhook url mint " + "failed agent_id=%s", + agent_id, + ) notify = "" if not notify: raise ValueError( @@ -1275,8 +1293,40 @@ async def get_job_status(self, job_id: str) -> Dict[str, Any]: body if isinstance(body, dict) else {"status": "unknown", "raw": body} ) except Exception as exc: + logger.exception( + "artifact_handler get_job_status failed job_id=%s", + jid, + ) return {"status": "unknown", "job_id": jid, "error": str(exc)} + async def confirm_artifact_imported(self, job_id: str) -> None: + """DELETE the retained jvforge artifact after a successful pull-import.""" + import httpx + + from jvagent.env import get_jvagent_jvforge_base_url + + jid = (job_id or "").strip() + if not jid: + return + forge_base = (get_jvagent_jvforge_base_url() or "").strip().rstrip("/") + if not forge_base: + return + url = f"{forge_base}/v1/artifacts/{jid}" + try: + async with httpx.AsyncClient(timeout=30.0) as client: + resp = await client.delete(url) + if resp.status_code not in (204, 404): + logger.error( + "artifact_handler artifact DELETE job_id=%s status=%s", + jid, + resp.status_code, + ) + except Exception: + logger.exception( + "artifact_handler artifact DELETE failed job_id=%s", + jid, + ) + # ── LLM tools (dispatched to custom_tools.py via VaultToolContext) ── def _load_custom_tools(self): diff --git a/jvagent/action/artifact_handler_interact_action/endpoints.py b/jvagent/action/artifact_handler_interact_action/endpoints.py index 79cdaaf3..aa085ddc 100644 --- a/jvagent/action/artifact_handler_interact_action/endpoints.py +++ b/jvagent/action/artifact_handler_interact_action/endpoints.py @@ -4,8 +4,9 @@ ``process_document_url`` when an async ingest job finishes. The vault downloads the artifact, imports the pageindex_graph into PageIndex, then sends a proactive notification (WhatsApp or Messenger) with a ready notice -and an optional answer. -(background, using call_model if there's a pending question). +and an optional answer. WhatsApp/Messenger send is scheduled with +``jvspatial.create_task`` (generate, then send); if a call returns a +scheduled object it is awaited before 200 so Lambda cannot freeze the send. On failure the endpoint returns 503 + Retry-After so jvforge retries the callback. On success it returns 200. @@ -16,11 +17,11 @@ import asyncio import json import logging -import time -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple from fastapi import Request from fastapi.responses import JSONResponse +from jvspatial import create_task from jvspatial.api import endpoint from jvspatial.api.endpoints.response import ResponseField, success_response @@ -41,6 +42,10 @@ logger = logging.getLogger(__name__) +_RETRY_AFTER_SECONDS = 30 +_READY_DESC_LOOKUP_TIMEOUT_S = 5.0 +_READY_GENERATE_TIMEOUT_S = 12.0 + async def _resolve_action(agent_id: str) -> Optional[Any]: """Resolve the ArtifactHandlerInteractAction instance for this agent. @@ -58,12 +63,38 @@ async def _resolve_action(agent_id: str) -> Optional[Any]: return None action = await agent.get_action_by_type("ArtifactHandlerInteractAction") if action is None: - pass + logger.error( + "artifact_handler_notify: ArtifactHandlerInteractAction missing " + "agent_id=%s", + agent_id, + ) return action except Exception: + logger.exception( + "artifact_handler_notify: action resolve failed agent_id=%s", + agent_id, + ) return None +async def _reload_action(action: Any) -> Any: + """Reload the action node from DB so lookup_job sees a persisted index.""" + action_id = getattr(action, "id", None) + if not action_id: + return action + try: + from jvagent.action.base import Action + + fresh = await Action.get(action_id) + return fresh if fresh is not None else action + except Exception: + logger.exception( + "artifact_handler_notify: action reload failed action_id=%s", + action_id, + ) + return action + + def _display_doc_name(entry: Dict[str, Any], payload_doc_name: str) -> str: """Prefer original filename; else strip ``{user_id}_`` from doc_name.""" filename = str(entry.get("filename") or "").strip() @@ -104,6 +135,7 @@ async def _doc_description_lookup( try: docs = await page_index.list_documents(access_control=False, summary=True) except Exception: + logger.exception("artifact_handler notify: desc lookup list_documents failed") return {} if not isinstance(docs, list): return {} @@ -118,6 +150,35 @@ async def _doc_description_lookup( return lookup +async def _await_ready_step(label: str, coro: Any, timeout: float, job_id: str) -> Any: + """Await a ready-message step; None on timeout or exception.""" + try: + return await asyncio.wait_for(coro, timeout=timeout) + except asyncio.TimeoutError: + logger.error( + "%s timed out job_id=%s timeout_s=%s", + label, + job_id, + timeout, + ) + return None + except Exception: + logger.exception( + "%s failed job_id=%s", + label, + job_id, + ) + return None + + +async def _await_create_task(coro: Any, *, name: str, job_id: str, kind: str) -> None: + """Shape B: run via ``create_task``; await a non-None scheduled return.""" + del job_id, kind + scheduled = await create_task(coro, name=name) + if scheduled is not None: + await scheduled + + async def _publish_whatsapp_message( *, agent: Any, @@ -139,6 +200,11 @@ async def _publish_whatsapp_message( """ memory = await agent.get_memory() if not memory: + logger.error( + "_publish_whatsapp_message: agent has no memory job_id=%s user_id=%s", + job_id, + user_id, + ) return False conversation = None @@ -148,21 +214,47 @@ async def _publish_whatsapp_message( conversation = await Conversation.get(conversation_id) except Exception: + logger.exception( + "_publish_whatsapp_message: Conversation.get failed job_id=%s " + "conversation_id=%s", + job_id, + conversation_id, + ) conversation = None if conversation is None: user = await memory.get_user(user_id, create_if_missing=False) if not user: + logger.error( + "_publish_whatsapp_message: user not found job_id=%s user_id=%s", + job_id, + user_id, + ) return False if session_id: conversation = await user.get_conversation_by_session(session_id) if conversation is None: + logger.error( + "_publish_whatsapp_message: conversation not found job_id=%s " + "user_id=%s session_id=%s conversation_id=%s", + job_id, + user_id, + session_id, + conversation_id, + ) return False effective_session_id = ( session_id or str(getattr(conversation, "session_id", "") or "").strip() or "" ) if not effective_session_id: + logger.error( + "_publish_whatsapp_message: no effective session_id job_id=%s " + "user_id=%s conversation_id=%s", + job_id, + user_id, + conversation_id, + ) return False interaction = await conversation.add_interaction( @@ -171,6 +263,13 @@ async def _publish_whatsapp_message( session_id=effective_session_id, ) if not interaction: + logger.error( + "_publish_whatsapp_message: add_interaction returned None job_id=%s " + "user_id=%s conversation_id=%s", + job_id, + user_id, + conversation_id, + ) return False interaction.add_parameter( @@ -203,17 +302,38 @@ async def _publish_whatsapp_message( whatsapp_action = await agent.get_action_by_type("WhatsAppAction") if whatsapp_action is None: + logger.error( + "_publish_whatsapp_message: WhatsAppAction missing job_id=%s user_id=%s", + job_id, + user_id, + ) return False try: if not whatsapp_action.is_configured(): + logger.error( + "_publish_whatsapp_message: WhatsAppAction not configured " + "job_id=%s user_id=%s", + job_id, + user_id, + ) return False except Exception: + logger.exception( + "_publish_whatsapp_message: is_configured failed job_id=%s user_id=%s", + job_id, + user_id, + ) return False try: api = await whatsapp_action.api() except Exception: + logger.exception( + "_publish_whatsapp_message: api() failed job_id=%s user_id=%s", + job_id, + user_id, + ) return False try: @@ -221,8 +341,22 @@ async def _publish_whatsapp_message( phone=user_id, message=content, ) - return isinstance(result, dict) and bool(result.get("ok", True)) + ok = isinstance(result, dict) and bool(result.get("ok", True)) + if not ok: + logger.error( + "_publish_whatsapp_message: send_message not ok job_id=%s " + "user_id=%s result=%r", + job_id, + user_id, + result, + ) + return ok except Exception: + logger.exception( + "_publish_whatsapp_message: send_message failed job_id=%s user_id=%s", + job_id, + user_id, + ) return False @@ -248,7 +382,7 @@ async def _publish_messenger_message( """ memory = await agent.get_memory() if not memory: - logger.warning("_publish_messenger_message: agent has no memory, cannot send") + logger.error("_publish_messenger_message: agent has no memory, cannot send") return False conversation = None @@ -263,14 +397,14 @@ async def _publish_messenger_message( if conversation is None: user = await memory.get_user(user_id, create_if_missing=False) if not user: - logger.warning( + logger.error( "_publish_messenger_message: user not found user_id=%s", user_id ) return False if session_id: conversation = await user.get_conversation_by_session(session_id) if conversation is None: - logger.warning( + logger.error( "_publish_messenger_message: conversation not found " "user_id=%s session_id=%s conversation_id=%s", user_id, @@ -283,7 +417,7 @@ async def _publish_messenger_message( session_id or str(getattr(conversation, "session_id", "") or "").strip() or "" ) if not effective_session_id: - logger.warning( + logger.error( "_publish_messenger_message: no effective session_id " "user_id=%s conversation_id=%s", user_id, @@ -297,7 +431,7 @@ async def _publish_messenger_message( session_id=effective_session_id, ) if not interaction: - logger.warning( + logger.error( "_publish_messenger_message: add_interaction returned None " "user_id=%s conversation_id=%s", user_id, @@ -338,20 +472,17 @@ async def _publish_messenger_message( try: response_bus = await agent.get_response_bus() except Exception: - logger.warning( - "_publish_messenger_message: get_response_bus failed", - exc_info=True, - ) + logger.exception("_publish_messenger_message: get_response_bus failed") return False if not response_bus: - logger.warning("_publish_messenger_message: no response bus") + logger.error("_publish_messenger_message: no response bus") return False adapter = response_bus._channel_adapters.get("messenger") if not adapter or not getattr(adapter, "_initialized", False): facebook_action = await agent.get_action_by_type("FacebookAction") if facebook_action is None: - logger.warning( + logger.error( "_publish_messenger_message: FacebookAction not found on agent" ) return False @@ -359,43 +490,36 @@ async def _publish_messenger_message( await facebook_action.ensure_page_access_token() await facebook_action.ensure_adapter_registered() except Exception: - logger.warning( - "_publish_messenger_message: ensure adapter/token failed", - exc_info=True, - ) + logger.exception("_publish_messenger_message: ensure adapter/token failed") return False adapter = response_bus._channel_adapters.get("messenger") if not adapter: - logger.warning( - "_publish_messenger_message: MessengerAdapter not registered" - ) + logger.error("_publish_messenger_message: MessengerAdapter not registered") return False facebook_action = getattr(adapter, "action", None) if facebook_action is None: - logger.warning( + logger.error( "_publish_messenger_message: MessengerAdapter has no FacebookAction" ) return False try: if not facebook_action.is_configured(): - logger.warning("_publish_messenger_message: FacebookAction not configured") + logger.error("_publish_messenger_message: FacebookAction not configured") return False except Exception: - logger.warning( - "_publish_messenger_message: FacebookAction is_configured() failed", - exc_info=True, + logger.exception( + "_publish_messenger_message: FacebookAction is_configured() failed" ) return False try: api = facebook_action.api() except Exception: - logger.warning( + logger.exception( "_publish_messenger_message: FacebookAction.api() failed on " - "registered action", - exc_info=True, + "registered action" ) return False @@ -409,23 +533,16 @@ async def _publish_messenger_message( result.get("error"), ) return False - logger.info( - "_publish_messenger_message: sent to user_id=%s job_id=%s", user_id, job_id - ) return True except Exception: - logger.error( - "_publish_messenger_message: send_text_message exception for user_id=%s", + logger.exception( + "_publish_messenger_message: send_message failed job_id=%s user_id=%s", + job_id, user_id, - exc_info=True, ) return False -_PROCESSING_STATUSES = frozenset({"queued", "processing", "pending", "submitted"}) - -_RETRY_AFTER_SECONDS = 30 - _ARTIFACT_404_RETRIES = 6 _ARTIFACT_404_BACKOFF_S = (1.0, 2.0, 4.0, 8.0, 10.0, 5.0) @@ -451,9 +568,6 @@ async def _download_and_import_graph( fetch_url = rewrite_process_document_url_to_jvforge_base(process_document_url) trusted = is_trusted_jvforge_url(fetch_url) - if fetch_url != process_document_url: - - pass raw_bytes: Optional[bytes] = None for attempt in range(1, _ARTIFACT_404_RETRIES + 1): try: @@ -471,17 +585,26 @@ async def _download_and_import_graph( ] await asyncio.sleep(delay) continue + logger.error( + "artifact_handler import: artifact fetch failed attempt=%s/%s: %s", + attempt, + _ARTIFACT_404_RETRIES, + msg, + ) return None if not raw_bytes: + logger.error("artifact_handler import: empty artifact body") return None try: graph = json.loads(raw_bytes) except Exception: + logger.exception("artifact_handler import: artifact is not JSON") return None if not isinstance(graph, dict): + logger.error("artifact_handler import: artifact JSON is not an object") return None roots = graph.get("roots") @@ -532,11 +655,52 @@ async def _download_and_import_graph( try: await _import_documents(graph, purge=False, collection_name=agent_id) except Exception: + logger.exception( + "artifact_handler import: PageIndex import failed agent_id=%s", + agent_id, + ) return None return effective_name +async def _close_reverse_index_job(action: Any, job_id: str) -> None: + """Mark notified and drop the jvforge reverse-index entry.""" + if action is None or not job_id: + return + try: + await action.mark_notified(job_id) + await action.clear_job(job_id) + except Exception: + logger.exception( + "artifact_handler_notify: clear_job failed job_id=%s", + job_id, + ) + + +async def _notify_and_close_job( + send_coro: Any, + *, + action: Any, + job_id: str, + channel: str, + agent_id: str, +) -> bool: + """Await a channel send, then close the reverse-index job on success.""" + sent = bool(await send_coro) + if sent: + await _close_reverse_index_job(action, job_id) + return True + logger.error( + "artifact_handler_notify: %s send failed job_id=%s " + "agent_id=%s; leaving reverse-index job for retry", + channel, + job_id, + agent_id, + ) + return False + + @endpoint( "/artifact_handler_action/notify/{agent_id}", methods=["POST"], @@ -574,8 +738,12 @@ async def artifact_handler_notify(request: Request, agent_id: str): 2. Require a known ``job_id`` in the reverse index (blocks replay/spam import). 3. Download artifact from ``process_document_url`` and import into PageIndex. 4. Mark the job as ``ready`` in conversation ``pending_ingest_jobs``. - 5. For WhatsApp/Messenger: send ready notice + optional answer. + 5. For WhatsApp/Messenger: two sequential ``create_task`` calls + (generate, then send). If a call returns a scheduled object it is + awaited before 200 so Lambda cannot freeze the send. + ``mark_notified`` / ``clear_job`` run only after a successful send. 6. Return 200 on success, 503 + Retry-After on failure (so jvforge retries). + A failed channel send leaves the reverse-index job in place. """ import hmac @@ -628,6 +796,11 @@ async def artifact_handler_notify(request: Request, agent_id: str): or not api_key_id or not hmac.compare_digest(expected_key, api_key_id) ): + logger.error( + "artifact_handler_notify: API key not authorized agent_id=%s job_id=%s", + agent_id, + job_id, + ) return JSONResponse( status_code=403, content={"detail": "API key not authorized for this agent"}, @@ -643,17 +816,35 @@ async def artifact_handler_notify(request: Request, agent_id: str): headers={"Retry-After": str(_RETRY_AFTER_SECONDS)}, ) - # Job lookup BEFORE download/import — unknown / cleared jobs must not - # trigger expensive PageIndex writes (replay / forged callbacks). + action = await _reload_action(action) + + # Job lookup BEFORE download/import — unknown jobs must not trigger + # expensive PageIndex writes (replay / forged callbacks). 503 so jvforge + # retries when the reverse-index save is not yet visible on this Lambda. entry = await action.lookup_job(job_id) if not entry: + index = getattr(action, "jvforge_job_index", None) or {} + index_size = len(index) if isinstance(index, dict) else 0 + logger.error( + "artifact_handler_notify: unknown job_id=%s agent_id=%s index_size=%s", + job_id, + agent_id, + index_size, + ) return JSONResponse( - status_code=404, - content={"detail": "unknown or already-cleared job_id"}, + status_code=503, + content={"detail": "unknown job_id"}, + headers={"Retry-After": str(_RETRY_AFTER_SECONDS)}, ) entry_agent = str(entry.get("agent_id") or "").strip() if entry_agent and entry_agent != agent_id: + logger.error( + "artifact_handler_notify: job_id=%s belongs to agent_id=%s not %s", + job_id, + entry_agent, + agent_id, + ) return JSONResponse( status_code=403, content={"detail": "job_id does not belong to this agent"}, @@ -670,6 +861,11 @@ async def artifact_handler_notify(request: Request, agent_id: str): imported_doc_name = await _download_and_import_graph(process_document_url, agent_id) if not imported_doc_name: + logger.error( + "artifact_handler_notify: graph import failed job_id=%s agent_id=%s", + job_id, + agent_id, + ) return JSONResponse( status_code=503, content={"detail": "graph import failed"}, @@ -680,13 +876,6 @@ async def artifact_handler_notify(request: Request, agent_id: str): session_id = str(entry.get("session_id") or "").strip() conversation_id = str(entry.get("conversation_id") or "").strip() channel = str(entry.get("channel") or "").strip().lower() or "default" - logger.info( - "artifact_handler_notify: job_id=%s channel=%s user_id=%s doc=%s", - job_id, - channel, - user_id, - doc_name, - ) # Prefer PageIndex import name; fall back to vault job name normalized the # same way PageIndex does (strip_redundant_md_suffix). vault_doc_name = str(entry.get("doc_name") or doc_name or "").strip() @@ -708,120 +897,220 @@ async def artifact_handler_notify(request: Request, agent_id: str): try: from jvagent.memory.conversation import Conversation + from .job_status import PROCESSING_STATUSES, apply_ingest_job_status + conv = await Conversation.get(conversation_id) if conv is not None: + pending = {} ctx = getattr(conv, "context", None) if isinstance(ctx, dict): vault = ctx.get("artifact_handler") if isinstance(vault, dict): - pending = vault.get("pending_ingest_jobs") - if isinstance(pending, dict) and job_id in pending: - job_entry = pending[job_id] - if isinstance(job_entry, dict): - prev_status = str(job_entry.get("status") or "").lower() - if ( - prev_status in _PROCESSING_STATUSES - or prev_status == "" - ): - job_entry["status"] = "ready" - job_entry["ready_at"] = time.time() - if internal_doc_name: - job_entry["doc_name"] = internal_doc_name - vault["active_doc_name"] = internal_doc_name - await conv.update_context( - {"artifact_handler": vault} - ) + raw = vault.get("pending_ingest_jobs") + if isinstance(raw, dict): + pending = raw + prev = ( + pending.get(job_id) if isinstance(pending.get(job_id), dict) else {} + ) + prev_status = str(prev.get("status") or "").lower() + if prev_status in PROCESSING_STATUSES or prev_status == "": + ok = await apply_ingest_job_status( + conv, + job_id, + "ready", + doc_name=internal_doc_name or None, + ) + if not ok: + logger.error( + "artifact_handler_notify: mark-ready failed job_id=%s", + job_id, + ) except Exception: - pass + logger.exception( + "artifact_handler_notify: mark-ready failed job_id=%s", + job_id, + ) # ── Send proactive notifications. # WhatsApp and Messenger get push messages; web/default relies on # check_ingest_status polling (TODO: add web push in a future phase). - if user_id and channel == "whatsapp": - asyncio.create_task( - _send_whatsapp_notifications( - agent_id=agent_id, - job_id=job_id or "", - user_id=user_id, - session_id=session_id, - conversation_id=conversation_id, - internal_doc_name=internal_doc_name, - display_doc=display_doc, - pending_question=pending_question, + # Two sequential Shape B tasks (generate, then send). Await a non-None + # create_task return so Lambda cannot 200-and-freeze the send. Close the + # reverse-index job only after a successful send. + notified = False + if user_id and channel in ("whatsapp", "messenger"): + content_box: List[Tuple[str, bool]] = [] + + async def _generate() -> None: + content_box.append( + await _generate_ready_content( + agent_id=agent_id, + job_id=job_id or "", + internal_doc_name=internal_doc_name, + display_doc=display_doc, + pending_question=pending_question, + ) ) + + await _await_create_task( + _generate(), + name=f"artifact_handler_generate_{job_id}", + job_id=job_id or "", + kind="generate", ) - elif user_id and channel == "messenger": - asyncio.create_task( - _send_messenger_notifications( - agent_id=agent_id, - job_id=job_id or "", - user_id=user_id, - session_id=session_id, - conversation_id=conversation_id, - internal_doc_name=internal_doc_name, - display_doc=display_doc, - pending_question=pending_question, + if content_box: + content, answered = content_box[0] + else: + content = _canned_ready_message( + display_doc, + pending_question=pending_question or None, ) + answered = False + + sent_box: List[bool] = [] + + async def _send() -> None: + from jvagent.core.agent import Agent + + try: + agent = await Agent.get(agent_id) + if agent is None: + logger.error( + "artifact_handler_notify: agent not found for send " + "agent_id=%s job_id=%s", + agent_id, + job_id, + ) + sent_box.append(False) + return + if channel == "whatsapp": + send_coro = _publish_whatsapp_message( + agent=agent, + user_id=user_id, + session_id=session_id, + conversation_id=conversation_id, + content=content, + display_doc=display_doc, + job_id=job_id or "", + answered=answered, + internal_doc_name=internal_doc_name, + pending_question=pending_question, + ) + else: + send_coro = _publish_messenger_message( + agent=agent, + user_id=user_id, + session_id=session_id, + conversation_id=conversation_id, + content=content, + display_doc=display_doc, + job_id=job_id or "", + answered=answered, + internal_doc_name=internal_doc_name, + pending_question=pending_question, + ) + sent_box.append( + await _notify_and_close_job( + send_coro, + action=action, + job_id=job_id or "", + channel=channel, + agent_id=agent_id, + ) + ) + except Exception: + logger.exception( + "artifact_handler_notify: send failed job_id=%s", + job_id, + ) + sent_box.append(False) + + await _await_create_task( + _send(), + name=f"artifact_handler_{channel}_{job_id}", + job_id=job_id or "", + kind="send", ) + sent = bool(sent_box and sent_box[0]) + if not sent: + return JSONResponse( + status_code=503, + content={"detail": f"{channel} notify failed"}, + headers={"Retry-After": str(_RETRY_AFTER_SECONDS)}, + ) + notified = True + else: + await _close_reverse_index_job(action, job_id) - # ── Mark notified + clear from jvforge reverse index. - if action is not None and job_id: - try: - await action.mark_notified(job_id) - await action.clear_job(job_id) - except Exception: - pass return { "status": "imported", "job_id": job_id, - "notified": channel in ("whatsapp", "messenger") and bool(user_id), + "notified": notified, "doc_name": imported_doc_name, } -async def _send_whatsapp_notifications( +async def _generate_ready_content( *, agent_id: str, job_id: str, - user_id: str, - session_id: str, - conversation_id: str, internal_doc_name: str, display_doc: str, pending_question: str, -) -> None: - """Send a single WhatsApp notification: ready notice, or ready + answer.""" - try: - from jvagent.core.agent import Agent +) -> Tuple[str, bool]: + """Build ready-notice text: 5s desc lookup + 12s generate, else canned. + Returns ``(content, answered)``. ``answered`` is True only when generate + produced a message before timeout. + """ + from jvagent.core.agent import Agent + + try: agent = await Agent.get(agent_id) if agent is None: - return + logger.error( + "_generate_ready_content: agent not found agent_id=%s job_id=%s", + agent_id, + job_id, + ) + return ( + _canned_ready_message( + display_doc, + pending_question=pending_question or None, + ), + False, + ) action = await _resolve_action(agent_id) - single_entry = { "internal_doc_name": internal_doc_name, "display_doc": display_doc, "pending_question": pending_question, } - desc_lookup: Dict[str, str] = {} - try: - desc_lookup = await _doc_description_lookup(agent, [single_entry]) - except Exception: - pass + desc_lookup = await _await_ready_step( + "_generate_ready_content: desc lookup", + _doc_description_lookup(agent, [single_entry]), + _READY_DESC_LOOKUP_TIMEOUT_S, + job_id, + ) + if not isinstance(desc_lookup, dict): + desc_lookup = {} doc_description = desc_lookup.get(internal_doc_name, "") content: Optional[str] = None answered = False - if pending_question and internal_doc_name and action is not None: - content = await _generate_ready_message( - agent=agent, - vault_action=action, - internal_doc_name=internal_doc_name, - display_doc=display_doc, - utterance=pending_question, - doc_description=doc_description or None, + content = await _await_ready_step( + "_generate_ready_content: generate", + _generate_ready_message( + agent=agent, + vault_action=action, + internal_doc_name=internal_doc_name, + display_doc=display_doc, + utterance=pending_question, + doc_description=doc_description or None, + ), + _READY_GENERATE_TIMEOUT_S, + job_id, ) if content: answered = True @@ -832,29 +1121,22 @@ async def _send_whatsapp_notifications( doc_description=doc_description, pending_question=pending_question or None, ) - - await _publish_whatsapp_message( - agent=agent, - user_id=user_id, - session_id=session_id, - conversation_id=conversation_id, - content=content, - display_doc=display_doc, - job_id=job_id, - answered=answered, - internal_doc_name=internal_doc_name, - pending_question=pending_question, - ) + return content, answered except Exception: - logger.error( - "_send_whatsapp_notifications: unexpected error agent_id=%s " "job_id=%s", - agent_id, + logger.exception( + "_generate_ready_content: using canned ready message job_id=%s", job_id, - exc_info=True, + ) + return ( + _canned_ready_message( + display_doc, + pending_question=pending_question or None, + ), + False, ) -async def _send_messenger_notifications( +async def _send_whatsapp_notifications( *, agent_id: str, job_id: str, @@ -864,86 +1146,29 @@ async def _send_messenger_notifications( internal_doc_name: str, display_doc: str, pending_question: str, -) -> None: - """Send a single Messenger notification: ready notice, or ready + answer.""" - logger.info( - "_send_messenger_notifications: starting agent_id=%s job_id=%s " - "user_id=%s doc=%s", - agent_id, - job_id, - user_id, - display_doc, - ) - try: - from jvagent.core.agent import Agent - - agent = await Agent.get(agent_id) - if agent is None: - logger.warning( - "_send_messenger_notifications: agent not found agent_id=%s", - agent_id, - ) - return - - action = await _resolve_action(agent_id) - - single_entry = { - "internal_doc_name": internal_doc_name, - "display_doc": display_doc, - "pending_question": pending_question, - } - desc_lookup: Dict[str, str] = {} - try: - desc_lookup = await _doc_description_lookup(agent, [single_entry]) - except Exception: - pass - doc_description = desc_lookup.get(internal_doc_name, "") - - content: Optional[str] = None - answered = False - - if pending_question and internal_doc_name and action is not None: - content = await _generate_ready_message( - agent=agent, - vault_action=action, - internal_doc_name=internal_doc_name, - display_doc=display_doc, - utterance=pending_question, - doc_description=doc_description or None, - ) - if content: - answered = True - - if not content: - content = _canned_ready_message( - display_doc, - doc_description=doc_description, - pending_question=pending_question or None, - ) +) -> bool: + """Generate ready text then publish via WhatsApp (used by tests).""" + from jvagent.core.agent import Agent - logger.info( - "_send_messenger_notifications: publishing to user_id=%s " - "answered=%s content_len=%d", - user_id, - answered, - len(content) if content else 0, - ) - await _publish_messenger_message( - agent=agent, - user_id=user_id, - session_id=session_id, - conversation_id=conversation_id, - content=content, - display_doc=display_doc, - job_id=job_id, - answered=answered, - internal_doc_name=internal_doc_name, - pending_question=pending_question, - ) - except Exception: - logger.error( - "_send_messenger_notifications: unexpected error agent_id=%s job_id=%s", - agent_id, - job_id, - exc_info=True, - ) + content, answered = await _generate_ready_content( + agent_id=agent_id, + job_id=job_id, + internal_doc_name=internal_doc_name, + display_doc=display_doc, + pending_question=pending_question, + ) + agent = await Agent.get(agent_id) + if agent is None: + return False + return await _publish_whatsapp_message( + agent=agent, + user_id=user_id, + session_id=session_id, + conversation_id=conversation_id, + content=content, + display_doc=display_doc, + job_id=job_id, + answered=answered, + internal_doc_name=internal_doc_name, + pending_question=pending_question, + ) diff --git a/jvagent/action/artifact_handler_interact_action/job_status.py b/jvagent/action/artifact_handler_interact_action/job_status.py new file mode 100644 index 00000000..6f9e2b46 --- /dev/null +++ b/jvagent/action/artifact_handler_interact_action/job_status.py @@ -0,0 +1,119 @@ +"""Shared ingest-job status writer for conversation vault + pending jobs.""" + +from __future__ import annotations + +import logging +import time +from typing import Any, Dict, Optional + +logger = logging.getLogger(__name__) + +VAULT_CTX_KEY = "artifact_handler" +PENDING_JOBS_KEY = "pending_ingest_jobs" +ACTIVE_DOC_KEY = "active_doc_name" + +PROCESSING_STATUSES = frozenset({"queued", "processing", "pending", "submitted"}) +FAILED_JOB_STATUSES = frozenset( + {"failed", "error", "cancelled", "canceled", "webhook_failed"} +) +READY_STATUSES = frozenset({"ready", "ingested"}) + +_VAULT_META_KEYS = frozenset({PENDING_JOBS_KEY, ACTIVE_DOC_KEY}) + + +def _now_ts() -> int: + return int(time.time()) + + +async def apply_ingest_job_status( + conversation: Any, + job_id: str, + status: str, + *, + doc_name: Optional[str] = None, + extra: Optional[Dict[str, Any]] = None, +) -> bool: + """Set ``queued`` / ``ready`` / ``failed`` on pending jobs and vault entries. + + Updates ``pending_ingest_jobs[job_id]`` and any vault list entry that + matches ``job_id`` (or ``doc_name`` when the entry has no job_id). + Returns True when the conversation context was persisted. + """ + jid = str(job_id or "").strip() + new_status = str(status or "").strip().lower() + if conversation is None or not jid or not new_status: + return False + + ctx = getattr(conversation, "context", None) + if not isinstance(ctx, dict): + ctx = {} + vault = ctx.get(VAULT_CTX_KEY) + if not isinstance(vault, dict): + vault = {} + else: + vault = dict(vault) + + pending_raw = vault.get(PENDING_JOBS_KEY) + pending: Dict[str, Dict[str, Any]] = {} + if isinstance(pending_raw, dict): + for key, value in pending_raw.items(): + if isinstance(value, dict) and key: + pending[str(key)] = dict(value) + + entry = dict(pending.get(jid) or {}) + entry["status"] = new_status + name = str(doc_name or "").strip() + if name: + entry["doc_name"] = name + if extra: + for key, value in extra.items(): + if value is not None: + entry[key] = value + if new_status in READY_STATUSES: + entry.setdefault("ready_at", _now_ts()) + elif new_status in FAILED_JOB_STATUSES: + entry.setdefault("failed_at", _now_ts()) + pending[jid] = entry + vault[PENDING_JOBS_KEY] = pending + + match_name = name or str(entry.get("doc_name") or "").strip() + for key, value in list(vault.items()): + if key in _VAULT_META_KEYS or not isinstance(value, list): + continue + updated: list = [] + for item in value: + if not isinstance(item, dict): + updated.append(item) + continue + row = dict(item) + row_job = str(row.get("job_id") or "").strip() + row_name = str(row.get("doc_name") or "").strip() + matched = row_job == jid or ( + not row_job and match_name and row_name == match_name + ) + if matched: + row["status"] = new_status + row["job_id"] = jid + if match_name: + row["doc_name"] = match_name + if new_status in READY_STATUSES: + row.setdefault("ready_at", entry.get("ready_at")) + elif new_status in FAILED_JOB_STATUSES: + row.setdefault("failed_at", entry.get("failed_at")) + updated.append(row) + vault[key] = updated + + if new_status in READY_STATUSES and match_name: + vault[ACTIVE_DOC_KEY] = match_name + + try: + await conversation.update_context({VAULT_CTX_KEY: vault}) + except Exception: + logger.exception( + "artifact_handler apply_ingest_job_status: context update failed " + "job_id=%s status=%s", + jid, + new_status, + ) + return False + return True diff --git a/jvagent/action/artifact_handler_interact_action/ready_message.py b/jvagent/action/artifact_handler_interact_action/ready_message.py index b7e35bba..f87d518e 100644 --- a/jvagent/action/artifact_handler_interact_action/ready_message.py +++ b/jvagent/action/artifact_handler_interact_action/ready_message.py @@ -7,9 +7,12 @@ from __future__ import annotations +import logging import re from typing import Any, Dict, List, Optional, Tuple +logger = logging.getLogger(__name__) + _IMAGE_EXTENSIONS = frozenset( { ".png", @@ -468,6 +471,10 @@ async def _generate_ready_message( text = await call_model(vault_action, user_prompt, system_prompt) except Exception: + logger.exception( + "_generate_ready_message: call_model failed doc=%s", + internal_doc_name, + ) return None if not isinstance(text, str) or not text.strip(): diff --git a/jvagent/action/artifact_handler_interact_action/webhook_auth.py b/jvagent/action/artifact_handler_interact_action/webhook_auth.py index c91cb1a4..7707af1f 100644 --- a/jvagent/action/artifact_handler_interact_action/webhook_auth.py +++ b/jvagent/action/artifact_handler_interact_action/webhook_auth.py @@ -3,8 +3,9 @@ Inbound route: ``/api/artifact_handler_action/notify/{agent_id}``. Credentials are persisted on ``ArtifactHandlerInteractAction``. -Keys are scoped to the **exact** notify path for one agent (no trailing -``/*`` wildcard) so a leaked notify key cannot hit another agent's callback. +Keys are minted with ``ALLOWED_WEBHOOK_ENDPOINT_GLOB`` (``…/notify/*``), +same as Drive/WhatsApp/PageIndex. Cross-agent binding is the handler hmac +of ``notify_webhook_api_key_id``. """ from jvagent.action.utils.webhook_system_user import webhook_system_user_factory diff --git a/jvagent/action/pageindex/README.md b/jvagent/action/pageindex/README.md index ead57cae..28760bcf 100644 --- a/jvagent/action/pageindex/README.md +++ b/jvagent/action/pageindex/README.md @@ -38,7 +38,7 @@ This scales to large document bases without full-corpus scans. When the lexical - `assimilate_document()` – ingestion (programmatic); builds lexical index during persist - `search_documents()` – retrieval (programmatic) -- `PageIndexAction` – core graph action: ingest, `search`, list, delete, **jvforge LLM webhook URL** (`get_webhook_url` / `handle_webhook_payload`; legacy webhook path preserved for jvforge clients) +- `PageIndexAction` – core graph action: ingest, `search`, list, delete, **jvforge LLM webhook URL** (`get_webhook_url` / `handle_webhook_payload`; path `/api/pageindex/interact/webhook/{agent_id}`) - `lexical_index` – inverted index (tokenizer, ranking, index CRUD) - REST endpoints under `/pageindex/` - Orchestrator tools: `pageindex__search`, `pageindex__assimilate`, etc. diff --git a/jvagent/action/pageindex/endpoints.py b/jvagent/action/pageindex/endpoints.py index 2baa7d96..25eba3ff 100644 --- a/jvagent/action/pageindex/endpoints.py +++ b/jvagent/action/pageindex/endpoints.py @@ -1974,7 +1974,7 @@ async def import_documents_endpoint( @endpoint( - "/pageindex_retrieval_interact_action/interact/webhook/{agent_id}", + "/pageindex/interact/webhook/{agent_id}", methods=["POST"], webhook=True, auth=False, diff --git a/jvagent/action/pageindex/pageindex_action/pageindex_action.py b/jvagent/action/pageindex/pageindex_action/pageindex_action.py index f7c24dcc..7dc3492e 100644 --- a/jvagent/action/pageindex/pageindex_action/pageindex_action.py +++ b/jvagent/action/pageindex/pageindex_action/pageindex_action.py @@ -51,11 +51,10 @@ class PageIndexAction(Action): # AUDIT-actions XC-4: admin-facing pageindex routes under # /agents/{agent_id}/pageindex/. ~18 routes; per-agent grouping. - # The /pageindex_retrieval_interact_action/interact/webhook/{agent_id} - # webhook also lives here for ingestion callbacks. + # The /pageindex/interact/webhook/{agent_id} LLM webhook also lives here. additional_endpoint_path_templates: ClassVar[List[str]] = [ "/agents/{agent_id}/pageindex/", - "/pageindex_retrieval_interact_action/interact/webhook/{agent_id}", + "/pageindex/interact/webhook/{agent_id}", ] strategy: str = attribute( diff --git a/jvagent/action/pageindex/webhook_auth.py b/jvagent/action/pageindex/webhook_auth.py index f8de48e1..f0d64c9a 100644 --- a/jvagent/action/pageindex/webhook_auth.py +++ b/jvagent/action/pageindex/webhook_auth.py @@ -1,14 +1,14 @@ """API key scope helper for PageIndex jvforge LLM webhook URLs. -Inbound route path remains ``.../pageindex_retrieval_interact_action/interact/webhook/{agent_id}``; -credentials are persisted on ``PageIndexAction``. +Inbound route: ``/api/pageindex/interact/webhook/{agent_id}``. +Credentials are persisted on ``PageIndexAction``. """ from jvagent.action.utils.webhook_system_user import webhook_system_user_factory SYSTEM_USER_EMAIL = "pageindex-retrieval-interact-action-service@system.internal" -WEBHOOK_PERMISSION = "webhook:pageindex_retrieval_interact_action" -PAGEINDEX_WEBHOOK_ROUTE_PREFIX = "pageindex_retrieval_interact_action/interact/webhook" +WEBHOOK_PERMISSION = "webhook:pageindex" +PAGEINDEX_WEBHOOK_ROUTE_PREFIX = "pageindex/interact/webhook" ALLOWED_WEBHOOK_ENDPOINT_GLOB = f"/api/{PAGEINDEX_WEBHOOK_ROUTE_PREFIX}/*" get_or_create_system_user = webhook_system_user_factory( diff --git a/jvagent/cli/server_config.py b/jvagent/cli/server_config.py index f0089675..98b82182 100644 --- a/jvagent/cli/server_config.py +++ b/jvagent/cli/server_config.py @@ -737,6 +737,12 @@ async def pre_startup_bootstrap( # Initialize all actions (channel adapters, TaskMonitor registration, …) await run_app_startup() + from jvagent.core.embed_endpoints import ( + remount_artifact_handler_notify_if_app_built, + ) + + remount_artifact_handler_notify_if_app_built(server) + # Ensure admin user exists admin_exists = await ensure_admin_user() diff --git a/jvagent/core/embed_endpoints.py b/jvagent/core/embed_endpoints.py index 0dbbe31f..10d21128 100644 --- a/jvagent/core/embed_endpoints.py +++ b/jvagent/core/embed_endpoints.py @@ -15,12 +15,141 @@ from __future__ import annotations +import logging +from typing import Any, List + +logger = logging.getLogger(__name__) + +_NOTIFY_PATH = "/artifact_handler_action/notify/{agent_id}" +_NOTIFY_PATH_MARKER = "artifact_handler_action/notify" + + +def _live_notify_routes(app: Any) -> List[str]: + """Return live FastAPI routes that look like the notify webhook.""" + found: List[str] = [] + for route in getattr(app, "routes", None) or []: + path = str(getattr(route, "path", "") or "") + if _NOTIFY_PATH_MARKER not in path: + continue + methods = sorted(str(m) for m in (getattr(route, "methods", None) or [])) + found.append(f"{path}[{','.join(methods)}]" if methods else path) + return found + def import_jvagent_endpoint_modules() -> None: """Import every first-party endpoint module in jvagent.""" from jvagent.action import endpoints as _action_endpoints # noqa: F401 + from jvagent.action.artifact_handler_interact_action import ( # noqa: F401 + endpoints as _ah_endpoints, + ) from jvagent.core import endpoints as _core_endpoints # noqa: F401 from jvagent.logging import endpoints as _logging_endpoints # noqa: F401 + cfg = ( + getattr( + _ah_endpoints.artifact_handler_notify, "_jvspatial_endpoint_config", None + ) + or {} + ) + logger.warning( + "artifact_handler notify: endpoints imported path=%s methods=%s " + "webhook=%s webhook_auth=%s", + cfg.get("path") or _NOTIFY_PATH, + list(cfg.get("methods") or ["POST"]), + cfg.get("webhook"), + cfg.get("webhook_auth"), + ) + + +def remount_artifact_handler_notify_if_app_built(server: Any) -> None: + """Mount notify on the live FastAPI app when ``get_app()`` already ran. + + Function ``@endpoint`` registration after ``get_app()`` updates the + registry only; Lambda/LWA then 404s. No-op when ``server.app`` is still + None (normal CLI: import happens before ``get_app()``). + """ + app = getattr(server, "app", None) + if app is None: + logger.warning( + "artifact_handler notify: remount skipped app_built=False " + "(route should land on later get_app())" + ) + return + live_before = _live_notify_routes(app) + remount = getattr(server, "_register_function_dynamically", None) + if not callable(remount): + logger.warning( + "artifact_handler notify: live FastAPI app already built but " + "Server has no _register_function_dynamically; notify may 404 " + "live_routes=%s", + live_before or "(none)", + ) + return + + from jvagent.action.artifact_handler_interact_action.endpoints import ( + artifact_handler_notify, + ) + + cfg = getattr(artifact_handler_notify, "_jvspatial_endpoint_config", None) or {} + registry = getattr(server, "_endpoint_registry", None) + info = None + if registry is not None: + getter = getattr(registry, "get_function_info", None) + if callable(getter): + info = getter(artifact_handler_notify) + + wrapped = None + path = _NOTIFY_PATH + methods = ["POST"] + if info is not None: + path = info.path or path + methods = list(info.methods or methods) or methods + route_config = (info.kwargs or {}).get("route_config") or {} + wrapped = route_config.get("endpoint") + else: + path = cfg.get("path") or path + methods = list(cfg.get("methods") or methods) or methods + + logger.warning( + "artifact_handler notify: remounting path=%s methods=%s " + "registry=%s live_routes_before=%s", + path, + methods, + info is not None, + live_before or "(none)", + ) + try: + remount( + wrapped or artifact_handler_notify, + path, + methods, + source_obj=artifact_handler_notify, + auth=cfg.get("auth_required", False), + permissions=cfg.get("permissions") or [], + roles=cfg.get("roles") or [], + response=cfg.get("response"), + webhook=cfg.get("webhook", True), + webhook_auth=cfg.get("webhook_auth", "api_key"), + ) + except Exception: + logger.warning( + "artifact_handler notify: failed to remount on live FastAPI app " + "path=%s live_routes=%s", + path, + _live_notify_routes(app) or "(none)", + exc_info=True, + ) + return + + live_after = _live_notify_routes(app) + logger.warning( + "artifact_handler notify: remount finished path=%s live_routes=%s", + path, + live_after or "(none)", + ) + -__all__ = ["import_jvagent_endpoint_modules"] +__all__ = [ + "import_jvagent_endpoint_modules", + "remount_artifact_handler_notify_if_app_built", +] diff --git a/jvagent/embed/bootstrap.py b/jvagent/embed/bootstrap.py index 636dae3f..48d64a19 100644 --- a/jvagent/embed/bootstrap.py +++ b/jvagent/embed/bootstrap.py @@ -257,6 +257,11 @@ def register_jvagent_endpoints_on_host( ) synced = sync_endpoint_modules(target_server) + from jvagent.core.embed_endpoints import ( + remount_artifact_handler_notify_if_app_built, + ) + + remount_artifact_handler_notify_if_app_built(target_server) logger.info("jvagent embed registered %d endpoint module(s) on host server", synced) return synced diff --git a/jvagent/skills/artifact_handler/scripts/custom_tools.py b/jvagent/skills/artifact_handler/scripts/custom_tools.py index 05e2a890..132f85a6 100644 --- a/jvagent/skills/artifact_handler/scripts/custom_tools.py +++ b/jvagent/skills/artifact_handler/scripts/custom_tools.py @@ -37,17 +37,21 @@ ``conversation.context["artifact_handler"]["active_doc_name"]``. On every ``ingest_document`` / ``list_my_documents`` / ``review_expired`` / ``check_ingest_status`` call, pending jobs are refreshed by checking -``PageIndexAction.list_documents`` with access control and newly-expired -docs are surfaced. +``PageIndexAction.list_documents`` and, when a doc is still missing, +jvforge job status (pull-import on ``webhook_failed`` / ``completed``). +Newly-expired docs are surfaced. """ from __future__ import annotations +import logging import re import time from typing import Any, Dict, List, Optional, Tuple from urllib.parse import urlparse +logger = logging.getLogger(__name__) + _SKILL_NAME = "artifact_handler" _IMAGE_EXTENSIONS = frozenset( @@ -662,13 +666,20 @@ async def _maybe_refresh_pending_jobs( session_id: str = "", user_id: str = "", ) -> Dict[str, Any]: - """Check pending async ingest jobs via PageIndex list_documents. + """Refresh pending ingest jobs via PageIndex, then jvforge pull-import. - Marks jobs ``ready`` when PageIndex lists the ``doc_name`` with - access-control filtering (so only docs accessible to this user/session - count). Surfaces a short ready message when a job transitions - (web/default backstop; WhatsApp usually already got the proactive ping). + Marks jobs ``ready`` when PageIndex lists the ``doc_name``, or when jvforge + reports ``webhook_failed`` / ``completed`` with an ``artifact_url`` and + pull-import succeeds. Marks ``failed`` when jvforge reports ``failed`` or + ``not_found``. Leaves jobs ``queued`` while jvforge is still processing + (including the notify delay, when ``artifact_url`` is hidden). """ + from jvagent.action.artifact_handler_interact_action.job_status import ( + FAILED_JOB_STATUSES, + READY_STATUSES, + apply_ingest_job_status, + ) + pending = _read_pending_jobs(conversation) if not pending: return {"refreshed": 0, "became_ready": [], "still_queued": [], "failed": []} @@ -691,7 +702,9 @@ def strip_redundant_md_suffix(name: str) -> str: # type: ignore[misc] str(d.get("doc_name") or "") for d in docs if isinstance(d, dict) } except Exception: - pass + logger.warning( + "artifact_handler refresh: list_documents failed", exc_info=True + ) def _resolve_available_name(doc_name: str) -> str: """Return the PageIndex name if doc_name or its md-stripped form is listed.""" @@ -705,24 +718,25 @@ def _resolve_available_name(doc_name: str) -> str: return stripped return "" - changed = False became_ready: List[str] = [] still_queued: List[str] = [] failed: List[str] = [] + dv_action = None for job_id, entry in list(pending.items()): status = str(entry.get("status") or "queued").lower() doc_name = str(entry.get("doc_name") or "") - if status in ("ready", "ingested"): - # Heal stale .md suffixes so search uses the PageIndex name. + if status in READY_STATUSES: resolved = _resolve_available_name(doc_name) if resolved and resolved != doc_name: + await apply_ingest_job_status( + conversation, job_id, "ready", doc_name=resolved + ) entry = dict(entry) entry["doc_name"] = resolved pending[job_id] = entry - changed = True continue - if status in _FAILED_JOB_STATUSES: + if status in FAILED_JOB_STATUSES: if doc_name: failed.append(doc_name) continue @@ -730,20 +744,48 @@ def _resolve_available_name(doc_name: str) -> str: resolved = _resolve_available_name(doc_name) if resolved: prev = status - entry = dict(entry) - entry["status"] = "ready" - entry["ready_at"] = _now_ts() - if resolved != doc_name: + ok = await apply_ingest_job_status( + conversation, job_id, "ready", doc_name=resolved + ) + if ok: + entry = dict(entry) + entry["status"] = "ready" entry["doc_name"] = resolved - pending[job_id] = entry - changed = True - if prev not in ("ready", "ingested") and resolved: - became_ready.append(resolved) - elif doc_name: + pending[job_id] = entry + if prev not in READY_STATUSES: + became_ready.append(resolved) + elif doc_name: + still_queued.append(doc_name) + continue + + if not doc_name: + continue + + if dv_action is None: + dv_action = await _get_artifact_handler_action(ctx) + if dv_action is None: + logger.info( + "artifact_handler refresh: PageIndex miss job_id=%s doc=%s " + "(no artifact_handler action; leaving queued)", + job_id, + doc_name, + ) still_queued.append(doc_name) + continue - if changed: - await _write_pending_jobs(conversation, pending) + outcome, ready_name = await _refresh_job_from_jvforge( + ctx, + conversation, + dv_action, + job_id, + doc_name, + ) + if outcome == "ready": + became_ready.append(ready_name or doc_name) + elif outcome == "failed": + failed.append(doc_name) + else: + still_queued.append(doc_name) if say_ready and became_ready: if len(became_ready) == 1: @@ -764,6 +806,88 @@ def _resolve_available_name(doc_name: str) -> str: } +async def _refresh_job_from_jvforge( + ctx: Any, + conversation: Any, + dv_action: Any, + job_id: str, + doc_name: str, +) -> Tuple[str, str]: + """Poll jvforge and pull-import when the push callback already finished. + + Returns ``(ready|failed|queued, doc_name)``. + """ + from jvagent.action.artifact_handler_interact_action.job_status import ( + PROCESSING_STATUSES, + apply_ingest_job_status, + ) + + forge = await dv_action.get_job_status(job_id) + forge_status = str((forge or {}).get("status") or "").strip().lower() + logger.info( + "artifact_handler refresh: job_id=%s jvforge_status=%s", + job_id, + forge_status or "unknown", + ) + if forge_status in PROCESSING_STATUSES or forge_status in ("", "unknown"): + return "queued", doc_name + if forge_status in ("failed", "not_found"): + await apply_ingest_job_status(conversation, job_id, "failed") + return "failed", doc_name + + artifact_url = str((forge or {}).get("artifact_url") or "").strip() + if forge_status not in ("webhook_failed", "completed") or not artifact_url: + logger.info( + "artifact_handler refresh: no pull-import job_id=%s status=%s " + "has_artifact_url=%s", + job_id, + forge_status, + bool(artifact_url), + ) + return "queued", doc_name + + from jvagent.action.artifact_handler_interact_action.endpoints import ( + _download_and_import_graph, + ) + + visitor = getattr(ctx, "visitor", None) + agent_id = _resolve_agent_id(ctx, visitor, dv_action) + if not agent_id: + logger.warning( + "artifact_handler refresh: pull-import skipped job_id=%s (no agent_id)", + job_id, + ) + return "queued", doc_name + + logger.info("artifact_handler refresh: pull-import start job_id=%s", job_id) + imported = await _download_and_import_graph(artifact_url, agent_id) + if not imported: + logger.warning("artifact_handler refresh: pull-import failed job_id=%s", job_id) + return "queued", doc_name + + ok = await apply_ingest_job_status(conversation, job_id, "ready", doc_name=imported) + if not ok: + logger.warning( + "artifact_handler refresh: mark-ready after pull-import failed job_id=%s", + job_id, + ) + return "queued", doc_name + try: + await dv_action.confirm_artifact_imported(job_id) + except Exception: + logger.warning( + "artifact_handler refresh: artifact DELETE failed job_id=%s", + job_id, + exc_info=True, + ) + logger.info( + "artifact_handler refresh: pull-import ok job_id=%s doc=%s", + job_id, + imported, + ) + return "ready", imported + + # ─── Tool 0: check_pending_attachments ──────────────────────────────── @@ -1661,8 +1785,9 @@ async def check_ingest_status(ctx) -> Dict[str, Any]: may still be queued. This is the status check during or after processing. Do not use ``check_pending_attachments`` for these questions. - Uses PageIndex list_documents with access control to verify a document - is available to the user, instead of polling jvforge for job status. + Uses PageIndex list_documents and jvforge job status (pull-import when + the notify callback failed) instead of treating missing PageIndex docs + as still queued forever. Also auto-runs (via helpers) on ingest_document / list_my_documents / review_expired activation. diff --git a/tests/action/artifact_handler_interact_action/test_artifact_handler_security.py b/tests/action/artifact_handler_interact_action/test_artifact_handler_security.py index 94617b6b..56488e41 100644 --- a/tests/action/artifact_handler_interact_action/test_artifact_handler_security.py +++ b/tests/action/artifact_handler_interact_action/test_artifact_handler_security.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch @@ -9,12 +10,15 @@ from jvspatial.api.exceptions import ValidationError from jvagent.action.artifact_handler_interact_action.artifact_handler_interact_action import ( + ArtifactHandlerInteractAction, _fetch_url_bytes_for_vault, ) from jvagent.action.artifact_handler_interact_action.endpoints import ( + _send_whatsapp_notifications, artifact_handler_notify, ) from jvagent.action.artifact_handler_interact_action.webhook_auth import ( + ALLOWED_WEBHOOK_ENDPOINT_GLOB, notify_endpoint_for_agent, ) @@ -23,6 +27,109 @@ def test_notify_endpoint_for_agent_is_exact_path(): path = notify_endpoint_for_agent("Agent:abc") assert path == "/api/artifact_handler_action/notify/Agent:abc" assert not path.endswith("*") + assert ALLOWED_WEBHOOK_ENDPOINT_GLOB == "/api/artifact_handler_action/notify/*" + + +def _notify_mint_patches(*, generate_key, get_key=None): + mock_service = MagicMock() + mock_service.generate_key = generate_key + mock_service.get_key = get_key or AsyncMock(return_value=None) + mock_service.revoke_key = AsyncMock() + return ( + patch.object( + ArtifactHandlerInteractAction, + "get_agent", + new_callable=AsyncMock, + return_value=SimpleNamespace(id="n.Agent.test123", name="TestAgent"), + ), + patch.object(ArtifactHandlerInteractAction, "save", new_callable=AsyncMock), + patch( + "jvagent.action.artifact_handler_interact_action.webhook_auth.get_or_create_system_user", + new_callable=AsyncMock, + return_value="o.User.system123", + ), + patch( + "jvspatial.api.auth.api_key_service.APIKeyService", + return_value=mock_service, + ), + patch("jvspatial.db.get_prime_database", return_value=MagicMock()), + patch("jvspatial.core.context.GraphContext", return_value=MagicMock()), + ), mock_service + + +@pytest.mark.asyncio +async def test_notify_webhook_mints_drive_style_glob(monkeypatch): + monkeypatch.setenv("JVAGENT_PUBLIC_BASE_URL", "http://localhost:8000") + action = ArtifactHandlerInteractAction() + mock_key = SimpleNamespace(id="o.APIKey.key123") + generate_key = AsyncMock(return_value=("test_mock_api_key", mock_key)) + patches, service = _notify_mint_patches(generate_key=generate_key) + with patches[0], patches[1], patches[2], patches[3], patches[4], patches[5]: + url = await action.get_notify_webhook_url() + assert url.startswith( + "http://localhost:8000/api/artifact_handler_action/notify/n.Agent.test123" + ) + assert "?api_key=test_mock_api_key" in url + kwargs = generate_key.call_args.kwargs + assert kwargs["allowed_endpoints"] == [ALLOWED_WEBHOOK_ENDPOINT_GLOB] + assert kwargs["permissions"] == ["webhook:artifact_handler_action"] + assert kwargs["allowed_ips"] == [] + service.generate_key.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_notify_webhook_remints_exact_path_only_key(monkeypatch): + monkeypatch.setenv("JVAGENT_PUBLIC_BASE_URL", "http://localhost:8000") + action = ArtifactHandlerInteractAction() + action.notify_webhook_url = ( + "http://localhost:8000/api/artifact_handler_action/notify/" + "n.Agent.test123?api_key=old" + ) + action.notify_webhook_api_key_id = "o.APIKey.old" + stale = SimpleNamespace( + is_active=True, + allowed_endpoints=["/api/artifact_handler_action/notify/n.Agent.test123"], + allowed_ips=[], + ) + mock_key = SimpleNamespace(id="o.APIKey.new") + generate_key = AsyncMock(return_value=("new_key", mock_key)) + patches, service = _notify_mint_patches( + generate_key=generate_key, + get_key=AsyncMock(return_value=stale), + ) + with patches[0], patches[1], patches[2], patches[3], patches[4], patches[5]: + url = await action.get_notify_webhook_url() + assert "new_key" in url + service.generate_key.assert_awaited_once() + assert service.generate_key.call_args.kwargs["allowed_endpoints"] == [ + ALLOWED_WEBHOOK_ENDPOINT_GLOB + ] + + +@pytest.mark.asyncio +async def test_notify_webhook_reuses_glob_scoped_key(monkeypatch): + monkeypatch.setenv("JVAGENT_PUBLIC_BASE_URL", "http://localhost:8000") + existing = ( + "http://localhost:8000/api/artifact_handler_action/notify/" + "n.Agent.test123?api_key=keep" + ) + action = ArtifactHandlerInteractAction() + action.notify_webhook_url = existing + action.notify_webhook_api_key_id = "o.APIKey.keep" + scoped = SimpleNamespace( + is_active=True, + allowed_endpoints=[ALLOWED_WEBHOOK_ENDPOINT_GLOB], + allowed_ips=[], + ) + generate_key = AsyncMock() + patches, service = _notify_mint_patches( + generate_key=generate_key, + get_key=AsyncMock(return_value=scoped), + ) + with patches[0], patches[1], patches[2], patches[3], patches[4], patches[5]: + url = await action.get_notify_webhook_url() + assert url == existing + service.generate_key.assert_not_awaited() @pytest.mark.asyncio @@ -56,6 +163,14 @@ def _request(*, api_key_id: str = "key-1", payload: dict | None = None): return req +async def _inline_create_task(coro_or_type, payload=None, **kwargs): + """Lambda Shape B: await the coroutine and return None.""" + if asyncio.iscoroutine(coro_or_type): + await coro_or_type + return None + raise AssertionError(f"expected a coroutine, got {type(coro_or_type)!r}") + + @pytest.mark.asyncio async def test_notify_rejects_missing_job_id(): req = _request(payload={"process_document_url": "https://example.com/a"}) @@ -96,6 +211,7 @@ async def test_notify_skips_import_for_unknown_job(): action = SimpleNamespace( notify_webhook_api_key_id="key-1", lookup_job=AsyncMock(return_value=None), + jvforge_job_index={}, ) req = _request( payload={ @@ -120,7 +236,8 @@ async def test_notify_skips_import_for_unknown_job(): ) as import_graph, ): resp = await artifact_handler_notify(req, "Agent:a") - assert resp.status_code == 404 + assert resp.status_code == 503 + assert resp.headers.get("Retry-After") import_graph.assert_not_awaited() @@ -158,7 +275,194 @@ async def test_notify_idempotent_when_already_notified(): "jvagent.action.artifact_handler_interact_action.endpoints._download_and_import_graph", new_callable=AsyncMock, ) as import_graph, + patch( + "jvagent.action.artifact_handler_interact_action.endpoints._publish_whatsapp_message", + new_callable=AsyncMock, + ) as send, ): out = await artifact_handler_notify(req, "Agent:a") assert out["status"] == "already_imported" import_graph.assert_not_awaited() + send.assert_not_awaited() + + +def _whatsapp_job(**extra): + entry = { + "job_id": "job-1", + "agent_id": "Agent:a", + "notified": False, + "user_id": "5926431530", + "session_id": "sess-1", + "conversation_id": "", + "channel": "whatsapp", + "doc_name": "upload.jpg", + "filename": "upload.jpg", + } + entry.update(extra) + return entry + + +def _notify_action(job_entry): + return SimpleNamespace( + notify_webhook_api_key_id="key-1", + lookup_job=AsyncMock(return_value=job_entry), + mark_notified=AsyncMock(), + clear_job=AsyncMock(), + jvforge_job_index={"job-1": job_entry}, + ) + + +@pytest.mark.asyncio +async def test_notify_awaits_whatsapp_send_before_clearing_job(): + action = _notify_action(_whatsapp_job()) + req = _request( + payload={ + "process_document_url": "https://example.com/a", + "job_id": "job-1", + "doc_name": "upload.jpg", + } + ) + send = AsyncMock(return_value=True) + with ( + patch( + "jvagent.action.artifact_handler_interact_action.endpoints._resolve_action", + new_callable=AsyncMock, + return_value=action, + ), + patch( + "jvagent.core.agent.Agent.get", + new_callable=AsyncMock, + return_value=SimpleNamespace(id="Agent:a"), + ), + patch( + "jvagent.action.artifact_handler_interact_action.endpoints._download_and_import_graph", + new_callable=AsyncMock, + return_value="5926431530_upload.jpg", + ), + patch( + "jvagent.action.artifact_handler_interact_action.endpoints._generate_ready_content", + new_callable=AsyncMock, + return_value=("Your image is ready. Ask me anything about it.", False), + ), + patch( + "jvagent.action.artifact_handler_interact_action.endpoints._publish_whatsapp_message", + send, + ), + patch( + "jvagent.action.artifact_handler_interact_action.endpoints.create_task", + _inline_create_task, + ), + ): + out = await artifact_handler_notify(req, "Agent:a") + assert out["status"] == "imported" + assert out["notified"] is True + send.assert_awaited_once() + action.mark_notified.assert_awaited_once_with("job-1") + action.clear_job.assert_awaited_once_with("job-1") + assert send.await_args.kwargs["user_id"] == "5926431530" + assert send.await_args.kwargs["job_id"] == "job-1" + + +@pytest.mark.asyncio +async def test_notify_returns_503_when_whatsapp_send_fails(): + action = _notify_action(_whatsapp_job()) + req = _request( + payload={ + "process_document_url": "https://example.com/a", + "job_id": "job-1", + } + ) + send = AsyncMock(return_value=False) + with ( + patch( + "jvagent.action.artifact_handler_interact_action.endpoints._resolve_action", + new_callable=AsyncMock, + return_value=action, + ), + patch( + "jvagent.core.agent.Agent.get", + new_callable=AsyncMock, + return_value=SimpleNamespace(id="Agent:a"), + ), + patch( + "jvagent.action.artifact_handler_interact_action.endpoints._download_and_import_graph", + new_callable=AsyncMock, + return_value="5926431530_upload.jpg", + ), + patch( + "jvagent.action.artifact_handler_interact_action.endpoints._generate_ready_content", + new_callable=AsyncMock, + return_value=("Your image is ready. Ask me anything about it.", False), + ), + patch( + "jvagent.action.artifact_handler_interact_action.endpoints._publish_whatsapp_message", + send, + ), + patch( + "jvagent.action.artifact_handler_interact_action.endpoints.create_task", + _inline_create_task, + ), + ): + resp = await artifact_handler_notify(req, "Agent:a") + assert resp.status_code == 503 + assert resp.headers.get("Retry-After") + send.assert_awaited_once() + action.mark_notified.assert_not_awaited() + action.clear_job.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_send_whatsapp_uses_canned_when_generate_times_out(): + action = SimpleNamespace(id="action-1") + agent = SimpleNamespace(id="Agent:a") + + async def _hang(**_kwargs): + await asyncio.sleep(30) + return "should not be used" + + publish = AsyncMock(return_value=True) + with ( + patch( + "jvagent.core.agent.Agent.get", + new_callable=AsyncMock, + return_value=agent, + ), + patch( + "jvagent.action.artifact_handler_interact_action.endpoints._resolve_action", + new_callable=AsyncMock, + return_value=action, + ), + patch( + "jvagent.action.artifact_handler_interact_action.endpoints._doc_description_lookup", + new_callable=AsyncMock, + return_value={}, + ), + patch( + "jvagent.action.artifact_handler_interact_action.endpoints._generate_ready_message", + _hang, + ), + patch( + "jvagent.action.artifact_handler_interact_action.endpoints._READY_GENERATE_TIMEOUT_S", + 0.05, + ), + patch( + "jvagent.action.artifact_handler_interact_action.endpoints._publish_whatsapp_message", + publish, + ), + ): + ok = await _send_whatsapp_notifications( + agent_id="Agent:a", + job_id="job-1", + user_id="5926431530", + session_id="sess-1", + conversation_id="conv-1", + internal_doc_name="upload.jpg", + display_doc="upload.jpg", + pending_question="what is this?", + ) + assert ok is True + publish.assert_awaited_once() + assert publish.await_args.kwargs["answered"] is False + content = publish.await_args.kwargs["content"] or "" + assert "ready" in content.lower() + assert "what is this?" in content diff --git a/tests/action/artifact_handler_interact_action/test_ingest_status_refresh.py b/tests/action/artifact_handler_interact_action/test_ingest_status_refresh.py new file mode 100644 index 00000000..940fb9a0 --- /dev/null +++ b/tests/action/artifact_handler_interact_action/test_ingest_status_refresh.py @@ -0,0 +1,312 @@ +"""Reverse-index persistence, vault status helper, and jvforge pull-import refresh.""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from jvagent.action.artifact_handler_interact_action.artifact_handler_interact_action import ( + ArtifactHandlerInteractAction, +) +from jvagent.action.artifact_handler_interact_action.job_status import ( + apply_ingest_job_status, +) + +_CUSTOM_TOOLS_PATH = ( + Path(__file__).resolve().parents[3] + / "jvagent" + / "skills" + / "artifact_handler" + / "scripts" + / "custom_tools.py" +) + + +def _load_custom_tools(): + spec = importlib.util.spec_from_file_location( + "artifact_handler_custom_tools_refresh_test", _CUSTOM_TOOLS_PATH + ) + assert spec is not None and spec.loader is not None + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +class FakeConversation: + def __init__(self, vault): + self.context = {"artifact_handler": vault} + + async def update_context(self, data): + for key, value in data.items(): + if key == "artifact_handler" and isinstance(value, dict): + vault = dict(self.context.get("artifact_handler") or {}) + vault.update(value) + self.context["artifact_handler"] = vault + else: + self.context[key] = value + + +def _action(*, save=None): + action = SimpleNamespace( + jvforge_job_index={}, + agent_id="n.Agent.a", + save=save or AsyncMock(), + ) + action._persist_job_index = ( + ArtifactHandlerInteractAction._persist_job_index.__get__(action) + ) + action.register_job = ArtifactHandlerInteractAction.register_job.__get__(action) + action.submit_ingest = ArtifactHandlerInteractAction.submit_ingest.__get__(action) + return action + + +@pytest.mark.asyncio +async def test_register_job_save_failure_raises(): + action = _action(save=AsyncMock(side_effect=RuntimeError("db down"))) + with pytest.raises(RuntimeError, match="db down"): + await action.register_job( + job_id="job-1", + user_id="u1", + conversation_id="c1", + session_id="s1", + channel="whatsapp", + doc_name="doc.jpg", + agent_id="n.Agent.a", + ) + + +@pytest.mark.asyncio +async def test_submit_ingest_fails_when_register_job_cannot_save(monkeypatch): + action = _action(save=AsyncMock(side_effect=RuntimeError("db down"))) + page_index = SimpleNamespace( + get_webhook_url=AsyncMock(return_value="https://example/llm") + ) + action.get_action = AsyncMock(return_value=page_index) + action.get_notify_webhook_url = AsyncMock(return_value="https://example/notify") + + monkeypatch.setattr( + "jvagent.env.get_jvagent_jvforge_base_url", + lambda: "https://forge.example", + ) + + async def fake_assimilate(**_kwargs): + return {"job_id": "job-1", "status": "queued"} + + monkeypatch.setattr( + "jvagent.action.pageindex.jvforge_assimilate.assimilate_via_jvforge_async", + fake_assimilate, + ) + with pytest.raises(RuntimeError, match="db down"): + await action.submit_ingest( + doc="https://files.example/a.jpg", + doc_name="a.jpg", + user_id="u1", + conversation_id="c1", + session_id="s1", + channel="whatsapp", + ) + + +@pytest.mark.asyncio +async def test_apply_ingest_job_status_updates_pending_and_vault(): + conv = FakeConversation( + { + "pending_ingest_jobs": { + "job-1": {"doc_name": "user_a.jpg", "status": "queued"} + }, + "private_u1": [ + { + "doc_name": "user_a.jpg", + "job_id": "job-1", + "status": "queued", + "filename": "a.jpg", + } + ], + } + ) + ok = await apply_ingest_job_status(conv, "job-1", "ready", doc_name="user_a.jpg") + assert ok is True + vault = conv.context["artifact_handler"] + assert vault["pending_ingest_jobs"]["job-1"]["status"] == "ready" + assert vault["private_u1"][0]["status"] == "ready" + assert vault["active_doc_name"] == "user_a.jpg" + + +@pytest.mark.asyncio +async def test_refresh_pageindex_hit_marks_ready(monkeypatch): + ct = _load_custom_tools() + conv = FakeConversation( + { + "pending_ingest_jobs": { + "job-1": {"doc_name": "user_a.jpg", "status": "queued"} + }, + "private_u1": [ + {"doc_name": "user_a.jpg", "job_id": "job-1", "status": "queued"} + ], + } + ) + page_index = SimpleNamespace( + list_documents=AsyncMock(return_value=[{"doc_name": "user_a.jpg"}]) + ) + interview = SimpleNamespace(get_action=AsyncMock(return_value=page_index)) + ctx = SimpleNamespace( + interview=interview, + visitor=SimpleNamespace(user_id="u1"), + add_directive=lambda _msg: None, + ) + + result = await ct._maybe_refresh_pending_jobs( + ctx, conv, say_ready=False, session_id="s1", user_id="u1" + ) + assert result["became_ready"] == ["user_a.jpg"] + assert result["still_queued"] == [] + assert ( + conv.context["artifact_handler"]["pending_ingest_jobs"]["job-1"]["status"] + == "ready" + ) + + +@pytest.mark.asyncio +async def test_refresh_webhook_failed_pull_imports(monkeypatch): + ct = _load_custom_tools() + conv = FakeConversation( + { + "pending_ingest_jobs": { + "job-1": {"doc_name": "user_a.jpg", "status": "queued"} + }, + "private_u1": [ + {"doc_name": "user_a.jpg", "job_id": "job-1", "status": "queued"} + ], + } + ) + page_index = SimpleNamespace(list_documents=AsyncMock(return_value=[])) + dv_action = SimpleNamespace( + agent_id="n.Agent.a", + get_job_status=AsyncMock( + return_value={ + "status": "webhook_failed", + "artifact_url": "https://forge.example/v1/artifacts/job-1", + } + ), + confirm_artifact_imported=AsyncMock(), + ) + + async def fake_get_action(name): + if name == "PageIndexAction": + return page_index + if name == "ArtifactHandlerInteractAction": + return dv_action + return None + + interview = SimpleNamespace(get_action=fake_get_action) + ctx = SimpleNamespace( + interview=interview, + visitor=SimpleNamespace(user_id="u1"), + add_directive=lambda _msg: None, + ) + + async def fake_import(url, agent_id): + assert "artifacts/job-1" in url + assert agent_id == "n.Agent.a" + return "user_a.jpg" + + monkeypatch.setattr( + "jvagent.action.artifact_handler_interact_action.endpoints._download_and_import_graph", + fake_import, + ) + + result = await ct._maybe_refresh_pending_jobs( + ctx, conv, say_ready=False, session_id="s1", user_id="u1" + ) + assert result["became_ready"] == ["user_a.jpg"] + assert result["still_queued"] == [] + assert ( + conv.context["artifact_handler"]["pending_ingest_jobs"]["job-1"]["status"] + == "ready" + ) + dv_action.confirm_artifact_imported.assert_awaited_once_with("job-1") + + +@pytest.mark.asyncio +async def test_refresh_jvforge_failed_marks_failed(): + ct = _load_custom_tools() + conv = FakeConversation( + { + "pending_ingest_jobs": { + "job-1": {"doc_name": "user_a.jpg", "status": "queued"} + } + } + ) + page_index = SimpleNamespace(list_documents=AsyncMock(return_value=[])) + dv_action = SimpleNamespace( + agent_id="n.Agent.a", + get_job_status=AsyncMock(return_value={"status": "failed"}), + confirm_artifact_imported=AsyncMock(), + ) + + async def fake_get_action(name): + if name == "PageIndexAction": + return page_index + if name == "ArtifactHandlerInteractAction": + return dv_action + return None + + ctx = SimpleNamespace( + interview=SimpleNamespace(get_action=fake_get_action), + visitor=SimpleNamespace(user_id="u1"), + add_directive=lambda _msg: None, + ) + result = await ct._maybe_refresh_pending_jobs( + ctx, conv, say_ready=False, session_id="s1", user_id="u1" + ) + assert result["failed"] == ["user_a.jpg"] + assert result["still_queued"] == [] + assert ( + conv.context["artifact_handler"]["pending_ingest_jobs"]["job-1"]["status"] + == "failed" + ) + dv_action.confirm_artifact_imported.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_refresh_processing_stays_queued(): + ct = _load_custom_tools() + conv = FakeConversation( + { + "pending_ingest_jobs": { + "job-1": {"doc_name": "user_a.jpg", "status": "queued"} + } + } + ) + page_index = SimpleNamespace(list_documents=AsyncMock(return_value=[])) + dv_action = SimpleNamespace( + agent_id="n.Agent.a", + get_job_status=AsyncMock(return_value={"status": "processing"}), + confirm_artifact_imported=AsyncMock(), + ) + + async def fake_get_action(name): + if name == "PageIndexAction": + return page_index + if name == "ArtifactHandlerInteractAction": + return dv_action + return None + + ctx = SimpleNamespace( + interview=SimpleNamespace(get_action=fake_get_action), + visitor=SimpleNamespace(user_id="u1"), + add_directive=lambda _msg: None, + ) + result = await ct._maybe_refresh_pending_jobs( + ctx, conv, say_ready=False, session_id="s1", user_id="u1" + ) + assert result["still_queued"] == ["user_a.jpg"] + assert result["became_ready"] == [] + assert ( + conv.context["artifact_handler"]["pending_ingest_jobs"]["job-1"]["status"] + == "queued" + ) diff --git a/tests/action/artifact_handler_interact_action/test_notify_route_mounted.py b/tests/action/artifact_handler_interact_action/test_notify_route_mounted.py new file mode 100644 index 00000000..0ce206a6 --- /dev/null +++ b/tests/action/artifact_handler_interact_action/test_notify_route_mounted.py @@ -0,0 +1,83 @@ +"""Notify webhook must be on the live FastAPI app (not registry-only).""" + +from __future__ import annotations + +import os + +from starlette.testclient import TestClient + +MINIMAL_APP_YAML = """ +app: jvagent_notify_route_test +context: + name: Notify Route Test + description: artifact_handler notify mount +config: + database: + type: json + path: ./test_jvdb + logging: + enabled: false + server: + host: 127.0.0.1 + port: 8766 +agents: [] +""" + +NOTIFY_PATH = "/api/artifact_handler_action/notify/n.Agent.test" + + +def _server(tmp_path, monkeypatch): + from jvagent.cli.server_config import ( + _set_db_env_from_config, + create_server_from_config, + ) + from jvagent.core.app_context import set_app_root + + monkeypatch.setenv( + "JVSPATIAL_JWT_SECRET_KEY", "test-jwt-secret-key-for-integration-tests" + ) + monkeypatch.setenv("JVSPATIAL_ENABLE_DEFERRED_SAVES", "false") + # Own DB env keys so _set_db_env_from_config cannot leak json into later tests. + monkeypatch.setenv("JVSPATIAL_DB_TYPE", "json") + monkeypatch.setenv("JVSPATIAL_DB_PATH", str(tmp_path / "test_jvdb")) + app_root = str(tmp_path) + (tmp_path / "app.yaml").write_text(MINIMAL_APP_YAML.strip(), encoding="utf-8") + set_app_root(app_root) + _set_db_env_from_config(app_root) + return create_server_from_config(debug=False, app_root=app_root) + + +def test_notify_route_on_live_app_after_create_server(tmp_path, monkeypatch): + """Eager import in create_server_from_config must land notify on get_app().""" + from jvagent.core.app import App + from jvagent.core.app_context import clear_app_root + + try: + server = _server(tmp_path, monkeypatch) + assert "JVSPATIAL_JSONDB_PATH" not in os.environ + client = TestClient(server.get_app()) + response = client.post(NOTIFY_PATH, json={}) + assert response.status_code != 404, response.text + finally: + App.clear_cache() + clear_app_root() + + +def test_notify_route_remount_if_app_already_built(tmp_path, monkeypatch): + """Late remount must keep POST notify off 404 when get_app() already ran.""" + from jvagent.core.app import App + from jvagent.core.app_context import clear_app_root + from jvagent.core.embed_endpoints import ( + remount_artifact_handler_notify_if_app_built, + ) + + try: + server = _server(tmp_path, monkeypatch) + app = server.get_app() + remount_artifact_handler_notify_if_app_built(server) + client = TestClient(app) + response = client.post(NOTIFY_PATH, json={}) + assert response.status_code != 404, response.text + finally: + App.clear_cache() + clear_app_root() diff --git a/tests/action/pageindex/test_pageindex_llm_webhook_url.py b/tests/action/pageindex/test_pageindex_llm_webhook_url.py new file mode 100644 index 00000000..14bc2add --- /dev/null +++ b/tests/action/pageindex/test_pageindex_llm_webhook_url.py @@ -0,0 +1,105 @@ +"""PageIndex LLM webhook URL is /api/pageindex/interact/webhook/{agent_id}.""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from jvagent.action.pageindex.pageindex_action.pageindex_action import PageIndexAction +from jvagent.action.pageindex.webhook_auth import ( + ALLOWED_WEBHOOK_ENDPOINT_GLOB, + PAGEINDEX_WEBHOOK_ROUTE_PREFIX, + WEBHOOK_PERMISSION, +) + +_MOD = "jvagent.action.pageindex.pageindex_action.pageindex_action" + + +def _mint_patches(*, generate_key): + mock_service = MagicMock() + mock_service.generate_key = generate_key + mock_service.get_key = AsyncMock(return_value=None) + mock_service.revoke_key = AsyncMock() + return ( + patch.object( + PageIndexAction, + "get_agent", + new_callable=AsyncMock, + return_value=SimpleNamespace(id="n.Agent.test123", name="TestAgent"), + ), + patch.object(PageIndexAction, "save", new_callable=AsyncMock), + patch(f"{_MOD}.get_public_base_url", return_value="http://localhost:8000"), + patch( + f"{_MOD}.get_or_create_system_user", + new_callable=AsyncMock, + return_value="o.User.system123", + ), + patch(f"{_MOD}.APIKeyService", return_value=mock_service), + patch(f"{_MOD}.get_prime_database", return_value=MagicMock()), + patch(f"{_MOD}.GraphContext", return_value=MagicMock()), + ), mock_service + + +def test_pageindex_webhook_path_constants(): + assert PAGEINDEX_WEBHOOK_ROUTE_PREFIX == "pageindex/interact/webhook" + assert ALLOWED_WEBHOOK_ENDPOINT_GLOB == "/api/pageindex/interact/webhook/*" + assert WEBHOOK_PERMISSION == "webhook:pageindex" + + +@pytest.mark.asyncio +async def test_pageindex_llm_webhook_mints_new_path(): + action = PageIndexAction.model_construct( + webhook_url=None, + webhook_api_key_id=None, + ) + mock_key = SimpleNamespace(id="o.APIKey.key123") + generate_key = AsyncMock(return_value=("test_mock_api_key", mock_key)) + patches, service = _mint_patches(generate_key=generate_key) + with ( + patches[0], + patches[1], + patches[2], + patches[3], + patches[4], + patches[5], + patches[6], + ): + url = await action.get_webhook_url() + assert url.startswith( + "http://localhost:8000/api/pageindex/interact/webhook/n.Agent.test123" + ) + assert "?api_key=test_mock_api_key" in url + kwargs = generate_key.call_args.kwargs + assert kwargs["allowed_endpoints"] == [ALLOWED_WEBHOOK_ENDPOINT_GLOB] + assert kwargs["permissions"] == [WEBHOOK_PERMISSION] + service.generate_key.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_pageindex_llm_webhook_remints_legacy_prefix(): + action = PageIndexAction.model_construct( + webhook_url=( + "http://localhost:8000/api/pageindex_retrieval_interact_action/" + "interact/webhook/n.Agent.test123?api_key=old" + ), + webhook_api_key_id="o.APIKey.old", + ) + mock_key = SimpleNamespace(id="o.APIKey.new") + generate_key = AsyncMock(return_value=("new_key", mock_key)) + patches, service = _mint_patches(generate_key=generate_key) + with ( + patches[0], + patches[1], + patches[2], + patches[3], + patches[4], + patches[5], + patches[6], + ): + url = await action.get_webhook_url() + assert "/api/pageindex/interact/webhook/n.Agent.test123" in url + assert "new_key" in url + service.generate_key.assert_awaited_once() + assert "pageindex_retrieval_interact_action" not in url diff --git a/tests/cli/test_server_config_postgres.py b/tests/cli/test_server_config_postgres.py index c44af86c..0cc2c803 100644 --- a/tests/cli/test_server_config_postgres.py +++ b/tests/cli/test_server_config_postgres.py @@ -7,12 +7,22 @@ """ import pytest +from jvspatial.api.config_groups import DatabaseConfig # Building a Server with db_type=postgres instantiates PostgresDB, which imports # asyncpg. It ships in the [test] extra; skip rather than error for anyone # running the suite without it. pytest.importorskip("asyncpg") +_DB_FIELDS = getattr(DatabaseConfig, "model_fields", None) or getattr( + DatabaseConfig, "__fields__", {} +) +if "postgres_dsn" not in _DB_FIELDS: + pytest.skip( + "jvspatial DatabaseConfig has no postgres fields (need >= 0.0.16)", + allow_module_level=True, + ) + POSTGRES_APP_YAML = """ app: pg_config_test context: @@ -79,6 +89,8 @@ def build_server(tmp_path, monkeypatch): monkeypatch.setenv("JVSPATIAL_JWT_SECRET_KEY", "test-secret-for-pg-config-tests") monkeypatch.setenv("JVAGENT_ADMIN_PASSWORD", "x") + monkeypatch.delenv("JVSPATIAL_DB_TYPE", raising=False) + monkeypatch.delenv("JVSPATIAL_DB_PATH", raising=False) for key in _PG_ENV: monkeypatch.delenv(key, raising=False) diff --git a/tests/integration/test_startup_health.py b/tests/integration/test_startup_health.py index f42898aa..62826a7b 100644 --- a/tests/integration/test_startup_health.py +++ b/tests/integration/test_startup_health.py @@ -40,6 +40,8 @@ async def test_pre_startup_bootstrap_admin_and_health(tmp_path, monkeypatch): "JVSPATIAL_JWT_SECRET_KEY", "test-jwt-secret-key-for-integration-tests" ) monkeypatch.setenv("JVSPATIAL_ENABLE_DEFERRED_SAVES", "false") + monkeypatch.setenv("JVSPATIAL_DB_TYPE", "json") + monkeypatch.setenv("JVSPATIAL_DB_PATH", str(tmp_path / "test_jvdb")) app_root = str(tmp_path) (tmp_path / "app.yaml").write_text(MINIMAL_APP_YAML.strip(), encoding="utf-8") @@ -90,6 +92,8 @@ async def test_bootstrap_only_creates_admin_without_preexisting_server( "JVSPATIAL_JWT_SECRET_KEY", "test-jwt-secret-key-for-integration-tests" ) monkeypatch.setenv("JVSPATIAL_ENABLE_DEFERRED_SAVES", "false") + monkeypatch.setenv("JVSPATIAL_DB_TYPE", "json") + monkeypatch.setenv("JVSPATIAL_DB_PATH", str(tmp_path / "test_jvdb")) app_root = str(tmp_path) (tmp_path / "app.yaml").write_text(MINIMAL_APP_YAML.strip(), encoding="utf-8")