feat: direct resource access - replace HTTP calls with Blob/Queue/Cosmos

- Add ContentProcessService for direct Azure resource access
- Fix queue message race condition: cancel renew task before delete
- Fix Cosmos insert ordering: create record before enqueuing
- Fix datetime serialization in gap_executor (default=str)
- Add interim status propagation via on_status_change callback
- Enable parallel document submission with asyncio.to_thread
- Add upsert_lock to prevent concurrent Cosmos write races
- Fix credential leak: replace config print with safe logger
- Fix queue client resource leak in close()
- ContentProcessor: write step status before execution for real-time UI
- Add unit tests for content_process_models and content_process_service
- Code quality: remove banner comments, dead code, debug prints

Author: Dongbumlee

src/ContentProcessor/src/libs/pipeline/queue_handler_base.py

@@ -15,7 +15,6 @@
 from abc import ABC, abstractmethod
 
 from azure.storage.queue import QueueClient
-
 from libs.application.application_context import AppContext
 from libs.base.application_models import AppModelBase
 from libs.models.content_process import ContentProcess, Step_Outputs
@@ -123,6 +122,29 @@ async def _connect_async(
 
                         self._current_message_context.data_pipeline.pipeline_status.active_step = self.handler_name
 
+                        # Update status to the currently running step BEFORE execution
+                        # so the UI reflects real-time progress.
+                        ContentProcess(
+                            process_id=self._current_message_context.data_pipeline.pipeline_status.process_id,
+                            processed_file_name=self._current_message_context.data_pipeline.files[
+                                0
+                            ].name,
+                            processed_file_mime_type=self._current_message_context.data_pipeline.files[
+                                0
+                            ].mime_type,
+                            status=step_name,
+                            imported_time=datetime.datetime.strptime(
+                                self._current_message_context.data_pipeline.pipeline_status.creation_time,
+                                "%Y-%m-%dT%H:%M:%S.%fZ",
+                            ),
+                            last_modified_time=datetime.datetime.now(datetime.UTC),
+                            last_modified_by=step_name,
+                        ).update_process_status_to_cosmos(
+                            connection_string=self.application_context.configuration.app_cosmos_connstr,
+                            database_name=self.application_context.configuration.app_cosmos_database,
+                            collection_name=self.application_context.configuration.app_cosmos_container_process,
+                        )
+
                         print(
                             f"Start Processing : {self.handler_name}"
                         ) if show_information else None

src/ContentProcessorWorkflow/src/libs/application/application_configuration.py

@@ -153,6 +153,15 @@ class Configuration(_configuration_base):
     )
     app_cps_processes: str = Field(default="cps-processes", alias="APP_CPS_PROCESSES")
 
+    app_cosmos_container_process: str = Field(
+        default="Processes", alias="APP_COSMOS_CONTAINER_PROCESS"
+    )
+    app_storage_blob_url: str = Field(default="", alias="APP_STORAGE_BLOB_URL")
+    app_storage_queue_url: str = Field(default="", alias="APP_STORAGE_QUEUE_URL")
+    app_message_queue_extract: str = Field(
+        default="content-pipeline-extract-queue", alias="APP_MESSAGE_QUEUE_EXTRACT"
+    )
+
     app_cps_content_process_endpoint: str = Field(
         default="http://localhost:8000/", alias="APP_CPS_CONTENT_PROCESS_ENDPOINT"
     )

src/ContentProcessorWorkflow/src/main.py

@@ -9,6 +9,7 @@
 """
 
 import asyncio
+import logging
 import os
 
 from sas.storage.blob.async_helper import AsyncStorageBlobHelper
@@ -22,8 +23,11 @@
 )
 from libs.base.application_base import ApplicationBase
 from repositories.claim_processes import Claim_Processes
+from services.content_process_service import ContentProcessService
 from steps.claim_processor import ClaimProcessor
 
+logger = logging.getLogger(__name__)
+
 
 class Application(ApplicationBase):
     """Local-development application that runs a single claim workflow.
@@ -40,10 +44,7 @@ def __init__(self):
 
     def initialize(self):
         """Bootstrap the application context and register services."""
-        print(
-            "Application initialized with configuration:",
-            self.application_context.configuration,
-        )
+        logger.info("Application initialized with configuration (secrets redacted)")
 
         self.register_services()
 
@@ -57,8 +58,9 @@ def register_services(self):
         )
 
         (
-            self.application_context
-            .add_singleton(DebuggingMiddleware, DebuggingMiddleware)
+            self.application_context.add_singleton(
+                DebuggingMiddleware, DebuggingMiddleware
+            )
             .add_singleton(LoggingFunctionMiddleware, LoggingFunctionMiddleware)
             .add_singleton(InputObserverMiddleware, InputObserverMiddleware)
             .add_singleton(Mem0AsyncMemoryManager, Mem0AsyncMemoryManager)
@@ -81,6 +83,13 @@ def register_services(self):
                     container_name=self.application_context.configuration.app_cosmos_container_batch_process,
                 ),
             )
+            .add_singleton(
+                ContentProcessService,
+                lambda: ContentProcessService(
+                    config=self.application_context.configuration,
+                    credential=self.application_context.credential,
+                ),
+            )
         )
 
     async def run(self):

src/ContentProcessorWorkflow/src/main_service.py

@@ -27,6 +27,7 @@
 )
 from libs.base.application_base import ApplicationBase
 from repositories.claim_processes import Claim_Processes
+from services.content_process_service import ContentProcessService
 from services.queue_service import (
     ClaimProcessingQueueService,
     QueueServiceConfig,
@@ -100,10 +101,7 @@ def initialize(self):
         Populates the DI container with agent-framework helpers, middlewares,
         repository services, and the queue-processing service.
         """
-        print(
-            "Application initialized with configuration:",
-            self.application_context.configuration,
-        )
+        logger.info("Application initialized with configuration (secrets redacted)")
         self.register_services()
 
     def register_services(self):
@@ -116,8 +114,9 @@ def register_services(self):
         )
 
         (
-            self.application_context
-            .add_singleton(DebuggingMiddleware, DebuggingMiddleware)
+            self.application_context.add_singleton(
+                DebuggingMiddleware, DebuggingMiddleware
+            )
             .add_singleton(LoggingFunctionMiddleware, LoggingFunctionMiddleware)
             .add_singleton(InputObserverMiddleware, InputObserverMiddleware)
             .add_singleton(Mem0AsyncMemoryManager, Mem0AsyncMemoryManager)
@@ -140,6 +139,13 @@ def register_services(self):
                     container_name=self.application_context.configuration.app_cosmos_container_batch_process,
                 ),
             )
+            .add_singleton(
+                ContentProcessService,
+                lambda: ContentProcessService(
+                    config=self.application_context.configuration,
+                    credential=get_azure_credential(),
+                ),
+            )
         )
 
         config = self._build_service_config(self.config_override)

src/ContentProcessorWorkflow/src/services/content_process_models.py

@@ -0,0 +1,86 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Pydantic models for content-processing queue messages.
+
+These models mirror the queue message schema expected by the
+ContentProcessor worker (``src/ContentProcessor``).  They must stay
+in sync with ``ContentProcessorAPI/app/routers/models/contentprocessor/model.py``.
+"""
+
+from datetime import datetime
+from enum import Enum
+from typing import Any, Optional
+
+from pydantic import BaseModel, ConfigDict, Field
+from sas.cosmosdb.mongo.model import RootEntityBase
+
+
+class ArtifactType(str, Enum):
+    Undefined = "undefined"
+    ConvertedContent = "converted_content"
+    ExtractedContent = "extracted_content"
+    SchemaMappedData = "schema_mapped_data"
+    ScoreMergedData = "score_merged_data"
+    SourceContent = "source_content"
+    SavedContent = "saved_content"
+
+
+class PipelineStep(str, Enum):
+    Transform = "transform"
+    Extract = "extract"
+    Mapping = "map"
+    Evaluating = "evaluate"
+    Save = "save"
+
+
+class ProcessFile(BaseModel):
+    process_id: str
+    id: str
+    name: str
+    size: int
+    mime_type: str
+    artifact_type: ArtifactType
+    processed_by: str
+
+
+class PipelineStatus(BaseModel):
+    process_id: str
+    schema_id: str
+    metadata_id: str
+    completed: Optional[bool] = Field(default=False)
+    creation_time: datetime
+    last_updated_time: Optional[datetime] = Field(default=None)
+    steps: list[str] = Field(default_factory=list)
+    remaining_steps: Optional[list[str]] = Field(default_factory=list)
+    completed_steps: Optional[list[str]] = Field(default_factory=list)
+
+
+class ContentProcessMessage(BaseModel):
+    """Queue message payload for a content-processing job."""
+
+    process_id: str
+    files: list[ProcessFile] = Field(default_factory=list)
+    pipeline_status: PipelineStatus = Field(default_factory=PipelineStatus)
+
+
+class ContentProcessRecord(RootEntityBase):
+    """Cosmos DB entity for the Processes collection.
+
+    Maps the document structure written by ContentProcessor/API.
+    Only the fields we read are declared; extra fields are allowed
+    so downstream writes (by the ContentProcessor worker) are preserved.
+    """
+
+    model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)
+
+    process_id: str = ""
+    processed_file_name: Optional[str] = None
+    processed_file_mime_type: Optional[str] = None
+    processed_time: Optional[str] = None
+    imported_time: Optional[datetime] = None
+    status: Optional[str] = None
+    entity_score: Optional[float] = 0.0
+    schema_score: Optional[float] = 0.0
+    result: Optional[Any] = None
+    confidence: Optional[Any] = None