Port main branch updates to refactored structure

Backport 13 commits from main into the refactored api/common/conserver
layout, adapting all file paths and imports from the old server/ structure.

Changes ported:
- diet link: S3 storage option for dialog bodies with presigned URLs,
  log redaction for sensitive options (aws_secret_access_key, etc.)
- wtf_transcribe link: new link for vfun GPU transcription server,
  produces WTF-format analysis entries
- webhook storage: new post-chain storage backend (parallel to existing
  conserver/links/webhook chain link)
- api: /stats/queue public endpoint for Redis list depth monitoring
- api: remove default VCON_REDIS_EXPIRY TTL on ingest; retention now
  controlled by storage backends
- api: index_vcon_parties() extracts party indexing from index_vcon()
  to avoid redundant Redis reads on ingest path
- tests: updated to reflect no-default-TTL and index_vcon_parties changes

Also fixes:
- pytest.ini pythonpath updated from old 'server' to 'common conserver api'
- pyproject.toml dev group: add httpx (required by fastapi TestClient)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: pavanputhra

api/api.py

@@ -250,7 +250,7 @@ async def version_endpoint() -> JSONResponse:
 )
 async def health_check() -> JSONResponse:
     """Health check endpoint.
-    
+
     Returns:
         JSONResponse with status and version info
     """
@@ -260,6 +260,24 @@ async def health_check() -> JSONResponse:
     })
 
 
+@app.get(
+    "/stats/queue",
+    summary="Get queue depth",
+    description="Returns the number of items in a Redis list (queue)",
+    tags=["system"],
+)
+async def get_queue_depth(
+    list_name: str = Query(..., description="Name of the Redis list to measure")
+) -> JSONResponse:
+    """Get the current depth of a Redis list. Public endpoint (no auth) for monitoring and backpressure."""
+    try:
+        depth = await redis_async.llen(list_name)
+        return JSONResponse(content={"list_name": list_name, "depth": depth})
+    except Exception as e:
+        logger.error(f"Error getting queue depth for '{list_name}': {str(e)}")
+        raise HTTPException(status_code=500, detail="Failed to get queue depth")
+
+
 class Vcon(BaseModel):
     """Pydantic model representing a vCon (Voice Conversation) record.
     
@@ -631,11 +649,6 @@ async def post_vcon(
     Stores the vCon in Redis and indexes it for searching. The vCon is added to a sorted
     set for timestamp-based retrieval and indexed by party information for searching.
     Optionally adds the vCon UUID to specified ingress lists for immediate processing.
-    
-    The vCon is stored with a default TTL of VCON_REDIS_EXPIRY seconds (default 3600s/1 hour).
-    This means vCons will automatically expire from Redis cache unless persisted to a
-    storage backend or the expiry is updated. Configure VCON_REDIS_EXPIRY environment
-    variable to change the default expiry time.
 
     Args:
         inbound_vcon: The vCon to store
@@ -659,16 +672,12 @@ async def post_vcon(
 
         logger.debug(f"Storing vCon {inbound_vcon.uuid} ({len(dict_vcon)} bytes)")
         await redis_async.json().set(key, "$", dict_vcon)
-        
-        # Set default expiry on newly created vCons
-        await redis_async.expire(key, VCON_REDIS_EXPIRY)
-        logger.debug(f"Set TTL of {VCON_REDIS_EXPIRY}s on vCon {inbound_vcon.uuid}")
-        
+
         logger.debug(f"Adding vCon {inbound_vcon.uuid} to sorted set")
         await add_vcon_to_set(key, timestamp)
 
         logger.debug(f"Indexing vCon {inbound_vcon.uuid}")
-        await index_vcon(inbound_vcon.uuid)
+        await index_vcon_parties(str(inbound_vcon.uuid), dict_vcon["parties"])
 
         # Add to ingress lists if specified
         if ingress_lists:
@@ -720,9 +729,7 @@ async def external_ingress_vcon(
     - Multiple API keys can be configured for the same ingress list
 
     The submitted vCon is stored, indexed, and automatically queued for processing
-    in the specified ingress list. The vCon is stored with a default TTL of 
-    VCON_REDIS_EXPIRY seconds (default 3600s/1 hour), after which it will expire
-    from Redis cache unless persisted to a storage backend.
+    in the specified ingress list.
 
     Args:
         request: FastAPI Request object for accessing headers
@@ -760,16 +767,12 @@ async def external_ingress_vcon(
             f"Storing vCon {inbound_vcon.uuid} ({len(dict_vcon)} bytes) via external ingress"
         )
         await redis_async.json().set(key, "$", dict_vcon)
-        
-        # Set default expiry on newly created vCons
-        await redis_async.expire(key, VCON_REDIS_EXPIRY)
-        logger.debug(f"Set TTL of {VCON_REDIS_EXPIRY}s on vCon {inbound_vcon.uuid}")
 
         logger.debug(f"Adding vCon {inbound_vcon.uuid} to sorted set")
         await add_vcon_to_set(key, timestamp)
 
         logger.debug(f"Indexing vCon {inbound_vcon.uuid}")
-        await index_vcon(inbound_vcon.uuid)
+        await index_vcon_parties(str(inbound_vcon.uuid), dict_vcon["parties"])
 
         # Always add to the specified ingress list (required for this endpoint)
         vcon_uuid_str = str(inbound_vcon.uuid)
@@ -1057,25 +1060,17 @@ async def get_dlq_vcons(
         raise HTTPException(status_code=500, detail="Failed to read DLQ")
 
 
-async def index_vcon(uuid: UUID) -> None:
-    """Index a vCon for searching.
+async def index_vcon_parties(vcon_uuid: str, parties: list) -> None:
+    """Index a vCon's parties for searching.
 
-    Adds the vCon to the sorted set and indexes it by party information
-    (tel, mailto, name) for searching. All indexed keys will expire after
-    VCON_INDEX_EXPIRY seconds.
+    Indexes by party information (tel, mailto, name). All indexed keys
+    will expire after VCON_INDEX_EXPIRY seconds.
 
     Args:
-        uuid: UUID of the vCon to index
+        vcon_uuid: UUID string of the vCon
+        parties: List of party dicts from the vCon
     """
-    key = f"vcon:{uuid}"
-    vcon = await redis_async.json().get(key)
-    created_at = datetime.fromisoformat(vcon["created_at"])
-    timestamp = int(created_at.timestamp())
-    vcon_uuid = vcon["uuid"]
-    await add_vcon_to_set(key, timestamp)
-
-    # Index by party information with expiration
-    for party in vcon["parties"]:
+    for party in parties:
         if party.get("tel"):
             tel_key = f"tel:{party['tel']}"
             await redis_async.sadd(tel_key, vcon_uuid)
@@ -1090,6 +1085,25 @@ async def index_vcon(uuid: UUID) -> None:
             await redis_async.expire(name_key, VCON_INDEX_EXPIRY)
 
 
+async def index_vcon(uuid: UUID) -> None:
+    """Index a vCon for searching (reads from Redis).
+
+    Reads the vCon from Redis, adds it to the sorted set, and indexes
+    by party information. Used for bulk re-indexing. For the ingest path,
+    use index_vcon_parties() directly to avoid redundant Redis reads.
+
+    Args:
+        uuid: UUID of the vCon to index
+    """
+    key = f"vcon:{uuid}"
+    vcon = await redis_async.json().get(key)
+    created_at = datetime.fromisoformat(vcon["created_at"])
+    timestamp = int(created_at.timestamp())
+    vcon_uuid = vcon["uuid"]
+    await add_vcon_to_set(key, timestamp)
+    await index_vcon_parties(vcon_uuid, vcon["parties"])
+
+
 @api_router.get(
     "/index_vcons",
     status_code=200,

common/storage/webhook/__init__.py

@@ -0,0 +1,46 @@
+import time
+
+from lib.vcon_redis import VconRedis
+from lib.logging_utils import init_logger
+from lib.metrics import record_histogram
+
+import requests
+
+logger = init_logger(__name__)
+
+default_options = {
+    "webhook-urls": [],
+    "headers": {},
+}
+
+
+def save(vcon_uuid, opts=default_options):
+    vcon_redis = VconRedis()
+    vCon = vcon_redis.get_vcon(vcon_uuid)
+
+    json_dict = vCon.to_dict()
+
+    if json_dict.get("vcon") == "0.0.1" or "vcon" not in json_dict:
+        json_dict["vcon"] = "0.3.0"
+
+    headers = opts.get("headers", {})
+
+    webhook_urls = opts.get("webhook-urls", [])
+    if not webhook_urls:
+        logger.warning(
+            f"webhook storage: no webhook-urls configured for vcon {vcon_uuid}, skipping"
+        )
+        return
+
+    for url in webhook_urls:
+        logger.info(
+            f"webhook storage: posting vcon {vcon_uuid} to webhook url: {url}"
+        )
+        webhook_start = time.time()
+        resp = requests.post(url, json=json_dict, headers=headers)
+        webhook_duration = round(time.time() - webhook_start, 3)
+        logger.info(
+            f"webhook storage response for {vcon_uuid}: {resp.status_code} {resp.text}"
+        )
+        record_histogram("conserver.webhook.duration", webhook_duration,
+                         attributes={"status_code": str(resp.status_code)})

common/tests/test_external_ingress.py

@@ -39,20 +39,21 @@ def test_validate_ingress_api_key_function(self):
 
     @patch("config.Configuration.get_ingress_auth")
     @patch("api.add_vcon_to_set")
-    @patch("api.index_vcon")
+    @patch("api.index_vcon_parties")
     def test_successful_submission_single_api_key(
-        self, mock_index_vcon, mock_add_vcon_to_set, mock_get_ingress_auth
+        self, mock_index_vcon_parties, mock_add_vcon_to_set, mock_get_ingress_auth
     ):
         """Test successful vCon submission with single API key configuration."""
         # Configure mocks
         mock_get_ingress_auth.return_value = {self.ingress_list: self.valid_api_key}
 
-        # Mock Redis client properly
+        # Mock Redis client properly (sadd/expire used by index_vcon_parties)
         mock_redis = MagicMock()
         mock_json = MagicMock()
         mock_json.set = AsyncMock()
         mock_redis.json.return_value = mock_json
         mock_redis.expire = AsyncMock()
+        mock_redis.sadd = AsyncMock()
         mock_redis.rpush = AsyncMock()
 
         # Set the global redis_async directly in the api module
@@ -77,35 +78,34 @@ def test_successful_submission_single_api_key(
 
             # Verify Redis operations were called
             mock_json.set.assert_called_once()
-            mock_redis.expire.assert_called_once()  # Verify expiry was set
             mock_redis.rpush.assert_called_once_with(
                 self.ingress_list, self.test_vcon["uuid"]
             )
             mock_add_vcon_to_set.assert_called_once()
-            mock_index_vcon.assert_called_once()
 
         finally:
             # Clean up the global variable
             api.redis_async = None
 
     @patch("config.Configuration.get_ingress_auth")
     @patch("api.add_vcon_to_set")
-    @patch("api.index_vcon")
+    @patch("api.index_vcon_parties")
     def test_successful_submission_multiple_api_keys(
-        self, mock_index_vcon, mock_add_vcon_to_set, mock_get_ingress_auth
+        self, mock_index_vcon_parties, mock_add_vcon_to_set, mock_get_ingress_auth
     ):
         """Test successful vCon submission with multiple API keys for same ingress."""
         # Configure mocks - multiple API keys for same ingress list
         mock_get_ingress_auth.return_value = {
             self.ingress_list: ["partner-1-key", self.valid_api_key, "partner-3-key"]
         }
 
-        # Mock Redis client properly
+        # Mock Redis client properly (sadd/expire used by index_vcon_parties)
         mock_redis = MagicMock()
         mock_json = MagicMock()
         mock_json.set = AsyncMock()
         mock_redis.json.return_value = mock_json
         mock_redis.expire = AsyncMock()
+        mock_redis.sadd = AsyncMock()
         mock_redis.rpush = AsyncMock()
 
         # Set the global redis_async directly in the api module
@@ -233,6 +233,7 @@ def test_redis_failure_handling(self, mock_get_ingress_auth):
         mock_json.set = AsyncMock(side_effect=Exception("Redis connection failed"))
         mock_redis.json.return_value = mock_json
         mock_redis.expire = AsyncMock()
+        mock_redis.sadd = AsyncMock()
         mock_redis.rpush = AsyncMock()
 
         # Set the global redis_async directly in the api module
@@ -254,9 +255,9 @@ def test_redis_failure_handling(self, mock_get_ingress_auth):
 
     @patch("config.Configuration.get_ingress_auth")
     @patch("api.add_vcon_to_set")
-    @patch("api.index_vcon")
+    @patch("api.index_vcon_parties")
     def test_multiple_ingress_lists_isolation(
-        self, mock_index_vcon, mock_add_vcon_to_set, mock_get_ingress_auth
+        self, mock_index_vcon_parties, mock_add_vcon_to_set, mock_get_ingress_auth
     ):
         """Test that API keys are properly isolated between ingress lists."""
         # Configure different API keys for different ingress lists
@@ -266,12 +267,13 @@ def test_multiple_ingress_lists_isolation(
             "shared_ingress": ["partner-a-key", "partner-b-key-1", "shared-key"],
         }
 
-        # Mock Redis client properly
+        # Mock Redis client properly (sadd/expire used by index_vcon_parties)
         mock_redis = MagicMock()
         mock_json = MagicMock()
         mock_json.set = AsyncMock()
         mock_redis.json.return_value = mock_json
         mock_redis.expire = AsyncMock()
+        mock_redis.sadd = AsyncMock()
         mock_redis.rpush = AsyncMock()
 
         # Set the global redis_async directly in the api module