✨(summary) taskV2 closer to target API gateway contract

Updated taskV2 API contract to be closer to the target gateway contract.
GET operations return the same things as the webhook payload.
Also store the summary on S3 to be iso with transcript.

src/helm/env.d/dev-keycloak/values.meet.yaml.gotmpl

@@ -6,7 +6,7 @@ _summaryEnvVars: &summaryEnvVars
   AWS_S3_ACCESS_KEY_ID: meet
   AWS_S3_SECRET_ACCESS_KEY: password
   AWS_S3_SECURE_ACCESS: False
-  AUTHORIZED_TENANTS: '[{"id": "dictaphone", "api_key": "dictaphone_token", "webhook_url": "http://dictaphone-backend.dictaphone.svc.cluster.local/api/v1.0/files/ai-webhook/", "webhook_api_key": "token_summary"}]'
+  AUTHORIZED_TENANTS: '[{"id": "dictaphone", "api_key": "dictaphone_token", "webhook_url": "http://dictaphone-backend.dictaphone.svc.cluster.local/api/v1.0/ai-jobs/webhook/", "webhook_api_key": "token_summary"}]'
   WHISPERX_API_KEY:
     secretKeyRef:
       name: secret-dev

src/summary/summary/api/route/tasks_v2.py

@@ -1,15 +1,24 @@
 """API routes related to application tasks (V2 / tenant friendly)."""
 
 from celery.result import AsyncResult
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Request
 
 from summary.core.celery_worker import (
+    celery,
     process_audio_transcribe_v2_task,
     summarize_v2_task,
 )
 from summary.core.config import AuthorizedTenant
 from summary.core.models import SummarizeTaskV2Request, TranscribeTaskV2Request
 from summary.core.security import verify_tenant_api_key_v2
+from summary.core.shared_models import (
+    SummarizeWebhookFailurePayload,
+    SummarizeWebhookPendingPayload,
+    SummarizeWebhookSuccessPayload,
+    TranscribeWebhookFailurePayload,
+    TranscribeWebhookPendingPayload,
+    TranscribeWebhookSuccessPayload,
+)
 
 router_tasks_v2 = APIRouter()
 
@@ -24,7 +33,7 @@ async def create_transcribe_task_v2(
         args=[{**request.model_dump(), "tenant_id": request_tenant.id}]
     )
 
-    return {"job_id": task.id, "message": "Transcribe job created"}
+    return TranscribeWebhookPendingPayload(job_id=task.id).model_dump()
 
 
 @router_tasks_v2.post("/async-jobs/summarize")
@@ -37,25 +46,64 @@ async def create_summarize_task_v2(
         args=[{**request.model_dump(), "tenant_id": request_tenant.id}]
     )
 
-    return {"job_id": task.id, "message": "Summarize job created"}
+    return SummarizeWebhookPendingPayload(job_id=task.id).model_dump()
 
 
 @router_tasks_v2.get("/async-jobs/transcribe/{job_id}")
-@router_tasks_v2.get("/async-jobs/summarize/{job_id}")
-async def get_task_status(
+async def get_transcribe_job_status(
     job_id: str,
+    request: Request,
     request_tenant: AuthorizedTenant = Depends(verify_tenant_api_key_v2),
 ):
-    """Check task status by ID."""
-    task = AsyncResult(job_id)
-    try:
-        if (
-            isinstance(task.args, (list, tuple))
-            and len(task.args) > 0
-            and isinstance(task.args[0], dict)
-            and task.args[0].get("tenant_id") == request_tenant.id
-        ):
-            return {"job_id": job_id, "status": task.status}
-    except (TypeError, KeyError):
-        pass
-    raise HTTPException(status_code=404, detail="Not found")
+    """Check transcription task status by ID."""
+    # We have to look directly in Redis to check if the task exists
+    redis_client = celery.backend.client
+    key = f"celery-task-meta-{job_id}"
+    if not redis_client.exists(key):
+        raise HTTPException(status_code=404, detail="Not found")
+
+    task = AsyncResult(job_id, app=celery)
+    task_tenant_id = task.args[0]["tenant_id"]
+    if task_tenant_id != request_tenant.id:
+        raise HTTPException(status_code=403, detail="Forbidden")
+
+    if task.status == "SUCCESS":
+        result = task.result
+        return TranscribeWebhookSuccessPayload.model_validate(result).model_dump()
+
+    if task.status == "FAILURE":
+        return TranscribeWebhookFailurePayload(
+            job_id=job_id, error_code="unknown_error"
+        ).model_dump()
+
+    return TranscribeWebhookPendingPayload(job_id=job_id).model_dump()
+
+
+@router_tasks_v2.get("/async-jobs/summarize/{job_id}")
+async def get_summarize_job_status(
+    job_id: str,
+    request: Request,
+    request_tenant: AuthorizedTenant = Depends(verify_tenant_api_key_v2),
+):
+    """Check summarize task status by ID."""
+    # We have to look directly in Redis to check if the task exists
+    redis_client = celery.backend.client
+    key = f"celery-task-meta-{job_id}"
+    if not redis_client.exists(key):
+        raise HTTPException(status_code=404, detail="Not found")
+
+    task = AsyncResult(job_id, app=celery)
+    task_tenant_id = task.args[0]["tenant_id"]
+    if task_tenant_id != request_tenant.id:
+        raise HTTPException(status_code=403, detail="Forbidden")
+
+    if task.status == "SUCCESS":
+        result = task.result
+        return SummarizeWebhookSuccessPayload.model_validate(result).model_dump()
+
+    if task.status == "FAILURE":
+        return SummarizeWebhookFailurePayload(
+            job_id=job_id, error_code="unknown_error"
+        ).model_dump()
+
+    return SummarizeWebhookPendingPayload(job_id=job_id).model_dump()

src/summary/summary/core/celery_worker.py

@@ -58,6 +58,9 @@ celery = Celery(
     broker=settings.celery_broker_url,
     backend=settings.celery_result_backend,
     broker_connection_retry_on_startup=True,
+    # To store the tasks args too in results and make the
+    # V2 API work
+    result_extended=True,
 )
 
 celery.config_from_object("summary.core.celery_config")
@@ -465,15 +468,14 @@ def process_audio_transcribe_v2_task(
         job_id=job_id,
     )
 
-    call_webhook_v2_task.apply_async(
-        args=[
-            TranscribeWebhookSuccessPayload(
-                job_id=job_id,
-                transcription_data_url=file_service.get_transcript_signed_url(job_id),
-            ).model_dump(),
-            payload.tenant_id,
-        ]
+    success_payload = TranscribeWebhookSuccessPayload(
+        job_id=job_id,
+        transcription_data_url=file_service.get_transcript_signed_url(job_id),
     )
+    call_webhook_v2_task.apply_async(
+        args=[success_payload.model_dump(), payload.tenant_id]
+    )
+    return success_payload.model_dump()
 
 
 @signals.task_failure.connect(sender=process_audio_transcribe_v2_task)
@@ -531,14 +533,17 @@ def summarize_v2_task(
         transcript=payload.content,
         session_id=self.request.id,
     )
-    call_webhook_v2_task.apply_async(
-        args=[
-            SummarizeWebhookSuccessPayload(
-                job_id=self.request.id, summary=summary
-            ).model_dump(),
-            payload.tenant_id,
-        ]
+    job_id = self.request.id
+    file_service.store_summary(summary=summary, job_id=job_id)
+
+    success_payload = SummarizeWebhookSuccessPayload(
+        job_id=job_id,
+        summary_data_url=file_service.get_summary_signed_url(job_id),
     )
+    call_webhook_v2_task.apply_async(
+        args=[success_payload.model_dump(), payload.tenant_id]
+    )
+    return success_payload.model_dump()
 
 
 @signals.task_failure.connect(sender=summarize_v2_task)

src/summary/summary/core/config.py

@@ -78,6 +78,7 @@ class Settings(BaseSettings):
     aws_s3_secret_access_key: SecretStr
     aws_s3_secure_access: bool = True
     aws_transcript_path: str = "transcripts"
+    aws_summary_path: str = "summaries"
 
     # AI-related settings
     whisperx_api_key: SecretStr

src/summary/summary/core/file_service.py

@@ -293,5 +293,27 @@ class FileService:
         transcript_path = f"{settings.aws_transcript_path}/{job_id}.json"
         logger.debug("Transcript path: %s", transcript_path)
         return self._minio_client.presigned_get_object(
-            self._bucket_name, transcript_path, expires=timedelta(hours=1)
+            self._bucket_name, transcript_path, expires=timedelta(hours=24)
+        )
+
+    def store_summary(self, *, summary: str, job_id: str) -> None:
+        """Store summary in MinIO."""
+        logger.info("Storing summary for job id %s", job_id)
+        summary_path = f"{settings.aws_summary_path}/{job_id}.txt"
+        logger.debug("Summary path: %s", summary_path)
+        data = summary.encode()
+        self._minio_client.put_object(
+            self._bucket_name,
+            summary_path,
+            io.BytesIO(data),
+            length=len(data),
+        )
+        logger.info("Summary stored successfully for job id %s", job_id)
+
+    def get_summary_signed_url(self, job_id: str) -> str:
+        """Get signed URL for summary file."""
+        summary_path = f"{settings.aws_summary_path}/{job_id}.txt"
+        logger.debug("Summary path: %s", summary_path)
+        return self._minio_client.presigned_get_object(
+            self._bucket_name, summary_path, expires=timedelta(hours=24)
         )

src/summary/summary/core/shared_models.py

@@ -65,6 +65,13 @@ class TranscribeWebhookSuccessPayload(BaseWebhook):
     )
 
 
+class TranscribeWebhookPendingPayload(BaseWebhook):
+    """Payload for a pending transcription webhook-like response."""
+
+    type: Literal["transcript"] = Field(default="transcript")
+    status: Literal["pending"] = Field(default="pending")
+
+
 class TranscribeWebhookFailurePayload(BaseWebhook):
     """Payload for a failed transcription webhook."""
 
@@ -76,7 +83,11 @@ class TranscribeWebhookFailurePayload(BaseWebhook):
 
 
 TranscribeWebhookPayloads = Annotated[
-    Union[TranscribeWebhookSuccessPayload, TranscribeWebhookFailurePayload],
+    Union[
+        TranscribeWebhookSuccessPayload,
+        TranscribeWebhookPendingPayload,
+        TranscribeWebhookFailurePayload,
+    ],
     Field(discriminator="status"),
 ]
 
@@ -86,7 +97,16 @@ class SummarizeWebhookSuccessPayload(BaseWebhook):
 
     type: Literal["summary"] = Field(default="summary")
     status: Literal["success"] = Field(default="success")
-    summary: str = Field(title="Summary", description="The summary of the text.")
+    summary_data_url: str = Field(
+        title="Summary", description="URL to the raw summary data."
+    )
+
+
+class SummarizeWebhookPendingPayload(BaseWebhook):
+    """Payload for a pending summarization webhook-like response."""
+
+    type: Literal["summary"] = Field(default="summary")
+    status: Literal["pending"] = Field(default="pending")
 
 
 class SummarizeWebhookFailurePayload(BaseWebhook):
@@ -100,7 +120,11 @@ class SummarizeWebhookFailurePayload(BaseWebhook):
 
 
 SummarizeWebhookPayloads = Annotated[
-    Union[SummarizeWebhookSuccessPayload, SummarizeWebhookFailurePayload],
+    Union[
+        SummarizeWebhookSuccessPayload,
+        SummarizeWebhookPendingPayload,
+        SummarizeWebhookFailurePayload,
+    ],
     Field(discriminator="status"),
 ]
 
@@ -114,8 +138,10 @@ webhook_payload_adapter = TypeAdapter(WebhookPayloads)
 
 __all__ = [
     "TranscribeWebhookSuccessPayload",
+    "TranscribeWebhookPendingPayload",
     "TranscribeWebhookFailurePayload",
     "SummarizeWebhookSuccessPayload",
+    "SummarizeWebhookPendingPayload",
     "SummarizeWebhookFailurePayload",
     "TranscribeWebhookPayloads",
     "SummarizeWebhookPayloads",

src/summary/tests/api/test_api_tasks_v2.py

@@ -1,6 +1,6 @@
 """Integration tests for the V2 task API endpoints."""
 
-from unittest.mock import MagicMock, patch
+from unittest.mock import ANY, MagicMock, patch
 
 
 class TestTasksV2:
@@ -26,7 +26,8 @@ class TestTasksV2:
         assert response.status_code == 200
         assert response.json() == {
             "job_id": "transcribe-task-id-abc",
-            "message": "Transcribe job created",
+            "type": "transcript",
+            "status": "pending",
         }
 
         args = mock_apply_async.call_args.kwargs["args"]
@@ -58,7 +59,8 @@ class TestTasksV2:
         assert response.status_code == 200
         assert response.json() == {
             "job_id": "summarize-task-id-abc",
-            "message": "Summarize job created",
+            "type": "summary",
+            "status": "pending",
         }
 
         args = mock_apply_async.call_args.kwargs["args"]
@@ -70,11 +72,13 @@ class TestTasksV2:
             }
         ]
 
+    @patch("summary.api.route.tasks_v2.celery")
     @patch("summary.api.route.tasks_v2.AsyncResult")
     def test_get_transcribe_task_status_returns_status_for_same_tenant(
-        self, mock_async_result, client
+        self, mock_async_result, mock_celery, client
     ):
         """GET /async-jobs/transcribe/{id} returns status when tenant matches."""
+        mock_celery.backend.client.exists.return_value = True
         mock_async_result.return_value = MagicMock(
             status="PENDING",
             args=[{"tenant_id": "test-tenant"}],
@@ -86,18 +90,28 @@ class TestTasksV2:
         )
 
         assert response.status_code == 200
-        assert response.json() == {"job_id": "task-id-abc", "status": "PENDING"}
+        assert response.json() == {
+            "job_id": "task-id-abc",
+            "type": "transcript",
+            "status": "pending",
+        }
 
-        mock_async_result.assert_called_once_with("task-id-abc")
+        mock_async_result.assert_called_once_with("task-id-abc", app=ANY)
 
+    @patch("summary.api.route.tasks_v2.celery")
     @patch("summary.api.route.tasks_v2.AsyncResult")
     def test_get_summarize_task_status_returns_status_for_same_tenant(
-        self, mock_async_result, client
+        self, mock_async_result, mock_celery, client
     ):
         """GET /async-jobs/summarize/{id} returns status when tenant matches."""
+        mock_celery.backend.client.exists.return_value = True
         mock_async_result.return_value = MagicMock(
             status="SUCCESS",
             args=[{"tenant_id": "test-tenant"}],
+            result={
+                "job_id": "task-id-abc",
+                "summary_data_url": "https://example.com/summary.json",
+            },
         )
 
         response = client.get(
@@ -106,16 +120,22 @@ class TestTasksV2:
         )
 
         assert response.status_code == 200
-        assert response.json() == {"job_id": "task-id-abc", "status": "SUCCESS"}
+        assert response.json() == {
+            "job_id": "task-id-abc",
+            "type": "summary",
+            "status": "success",
+            "summary_data_url": "https://example.com/summary.json",
+        }
 
-        mock_async_result.assert_called_once_with("task-id-abc")
+        mock_async_result.assert_called_once_with("task-id-abc", app=ANY)
 
+    @patch("summary.api.route.tasks_v2.celery")
     @patch("summary.api.route.tasks_v2.AsyncResult")
-    def test_get_task_status_returns_404_when_args_are_missing(
-        self, mock_async_result, client
+    def test_get_task_status_returns_404_when_job_does_not_exist(
+        self, mock_async_result, mock_celery, client
     ):
-        """GET /async-jobs/.../{id} returns 404 when task args are invalid."""
-        mock_async_result.return_value = MagicMock(status="SUCCESS", args=None)
+        """GET /async-jobs/.../{id} returns 404 when task key is not in Redis."""
+        mock_celery.backend.client.exists.return_value = False
 
         response = client.get(
             "/api/v2/async-jobs/transcribe/task-id-abc",
@@ -124,3 +144,24 @@ class TestTasksV2:
 
         assert response.status_code == 404
         assert response.json() == {"detail": "Not found"}
+        mock_async_result.assert_not_called()
+
+    @patch("summary.api.route.tasks_v2.celery")
+    @patch("summary.api.route.tasks_v2.AsyncResult")
+    def test_get_task_status_returns_403_for_other_tenant(
+        self, mock_async_result, mock_celery, client
+    ):
+        """GET /async-jobs/.../{id} returns 403 when task belongs to another tenant."""
+        mock_celery.backend.client.exists.return_value = True
+        mock_async_result.return_value = MagicMock(
+            status="PENDING",
+            args=[{"tenant_id": "another-tenant"}],
+        )
+
+        response = client.get(
+            "/api/v2/async-jobs/summarize/task-id-abc",
+            headers={"Authorization": "Bearer test-api-token"},
+        )
+
+        assert response.status_code == 403
+        assert response.json() == {"detail": "Forbidden"}