fix(artifact_handler): resolve webhook errors, await message delivery, and sync job statuses

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

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,17 +1108,33 @@ 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
         index = dict(self.jvforge_job_index or {})
         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):