Phase 7: Integrated Vectorized Long-Term Memory (Index on Approval, Retrieve on Run)

backend/routers/agent_runner.py

@@ -1,7 +1,14 @@
-from fastapi import APIRouter, HTTPException, BackgroundTasks
+from fastapi import APIRouter, HTTPException, BackgroundTasks, Request
+from fastapi.responses import StreamingResponse
 from services.supabase_service import supabase
 from services.agent_runner_service import AgentRunnerService
+from services.config import settings
+from services.audit_service import audit_service
 from services.output_quality import report_text_from_output
+from services.task_queue import TaskQueueService
+from services.memory_service import memory_service
+import asyncio
+import json
 import logging
 
 router = APIRouter()
@@ -25,7 +32,24 @@ def _assert_task_quality(task: dict):
     reasons = quality_review.get("fail_reasons") or ["Task output failed quality validation."]
     raise HTTPException(status_code=400, detail=f"Task output failed quality review: {'; '.join(reasons)}")
 
+def _assert_project_is_mutable(project_id: str):
+    project = supabase.table("projects").select("id,status").eq("id", project_id).single().execute().data
+    if not project:
+        raise HTTPException(status_code=404, detail="Project not found")
+    if project.get("status") == "completed":
+        raise HTTPException(status_code=409, detail="Completed projects are locked and cannot be modified.")
+
+def _assert_task_project_is_mutable(task: dict):
+    project_id = task.get("project_id")
+    if project_id:
+        _assert_project_is_mutable(project_id)
+
 def update_task_status(task_id: str, status: str):
+    task_res = supabase.table("tasks").select("project_id").eq("id", task_id).single().execute()
+    if not task_res.data:
+        raise HTTPException(status_code=404, detail="Task not found")
+    _assert_task_project_is_mutable(task_res.data)
+
     result = (
         supabase.table("tasks")
         .update({"status": status})
@@ -53,8 +77,218 @@ def update_task_status(task_id: str, status: str):
 
     return task_data
 
+
+def _sse_event(event: str, data: dict, event_id: str | None = None) -> str:
+    lines = []
+    if event_id:
+        lines.append(f"id: {event_id}")
+    lines.append(f"event: {event}")
+    payload = json.dumps(data, default=str)
+    for line in payload.splitlines() or ["{}"]:
+        lines.append(f"data: {line}")
+    return "\n".join(lines) + "\n\n"
+
+
+def _project_task_ids(project_id: str) -> list[str]:
+    rows = (
+        supabase.table("tasks")
+        .select("id")
+        .eq("project_id", project_id)
+        .execute()
+        .data
+        or []
+    )
+    return [row["id"] for row in rows if row.get("id")]
+
+
+def _user_id_from_access_token(access_token: str | None) -> str:
+    if not access_token:
+        raise HTTPException(status_code=401, detail="Missing access token")
+    try:
+        auth_user = supabase.auth.get_user(access_token)
+        user = getattr(auth_user, "user", None)
+        user_id = getattr(user, "id", None)
+        if not user_id and isinstance(auth_user, dict):
+            user_id = auth_user.get("user", {}).get("id")
+    except Exception as exc:
+        logger.warning("Could not validate log stream access token: %s", exc)
+        raise HTTPException(status_code=401, detail="Invalid access token") from exc
+    if not user_id:
+        raise HTTPException(status_code=401, detail="Invalid access token")
+    return user_id
+
+
+def _team_ids_for_user(user_id: str) -> list[str]:
+    try:
+        rows = (
+            supabase.table("team_members")
+            .select("team_id")
+            .eq("user_id", user_id)
+            .execute()
+            .data
+            or []
+        )
+    except Exception as exc:
+        logger.warning("Team membership lookup unavailable for log stream: %s", exc)
+        return []
+    return [row["team_id"] for row in rows if row.get("team_id")]
+
+
+def _project_ids_for_user(user_id: str) -> list[str]:
+    project_ids: set[str] = set()
+
+    owned = (
+        supabase.table("projects")
+        .select("id")
+        .eq("owner_id", user_id)
+        .execute()
+        .data
+        or []
+    )
+    project_ids.update(row["id"] for row in owned if row.get("id"))
+
+    public = (
+        supabase.table("projects")
+        .select("id")
+        .eq("is_public", True)
+        .execute()
+        .data
+        or []
+    )
+    project_ids.update(row["id"] for row in public if row.get("id"))
+
+    team_ids = _team_ids_for_user(user_id)
+    if team_ids:
+        team_projects = (
+            supabase.table("projects")
+            .select("id")
+            .in_("team_id", team_ids)
+            .execute()
+            .data
+            or []
+        )
+        project_ids.update(row["id"] for row in team_projects if row.get("id"))
+
+    return list(project_ids)
+
+
+def _can_view_project_for_user(project_id: str, user_id: str) -> bool:
+    if not project_id:
+        return False
+    if project_id in _project_ids_for_user(user_id):
+        return True
+    return False
+
+
+def _authorized_task_ids(user_id: str, project_id: str | None = None, task_id: str | None = None) -> list[str]:
+    if task_id:
+        task = supabase.table("tasks").select("id,project_id").eq("id", task_id).single().execute().data
+        if not task or not _can_view_project_for_user(task.get("project_id"), user_id):
+            raise HTTPException(status_code=403, detail="Task logs are not visible to this user")
+        return [task_id]
+
+    if project_id:
+        if not _can_view_project_for_user(project_id, user_id):
+            raise HTTPException(status_code=403, detail="Project logs are not visible to this user")
+        return _project_task_ids(project_id)
+
+    project_ids = _project_ids_for_user(user_id)
+    if not project_ids:
+        return []
+    rows = (
+        supabase.table("tasks")
+        .select("id")
+        .in_("project_id", project_ids)
+        .execute()
+        .data
+        or []
+    )
+    return [row["id"] for row in rows if row.get("id")]
+
+
+def _fetch_recent_logs(
+    limit: int = 50,
+    after_created_at: str | None = None,
+    *,
+    task_ids: list[str],
+) -> list[dict]:
+    if not task_ids:
+        return []
+    query = (
+        supabase.table("agent_logs")
+        .select("id,task_id,run_id,action,content,metadata,created_at")
+        .order("created_at", desc=after_created_at is None)
+        .limit(limit)
+        .in_("task_id", task_ids)
+    )
+    if after_created_at:
+        query = query.gt("created_at", after_created_at)
+    rows = query.execute().data or []
+    return rows if after_created_at else list(reversed(rows))
+
+
+@router.get("/logs/stream")
+async def stream_agent_logs(
+    request: Request,
+    limit: int = 50,
+    project_id: str | None = None,
+    task_id: str | None = None,
+    access_token: str | None = None,
+):
+    """
+    Streams agent log inserts as Server-Sent Events.
+    """
+    if project_id and task_id:
+        raise HTTPException(status_code=400, detail="Use either project_id or task_id, not both.")
+    user_id = _user_id_from_access_token(access_token)
+    task_ids = _authorized_task_ids(user_id, project_id=project_id, task_id=task_id)
+
+    async def event_generator():
+        last_created_at = None
+        sent_ids: set[str] = set()
+        yield _sse_event("ready", {
+            "message": "Agent log stream connected",
+            "project_id": project_id,
+            "task_id": task_id,
+            "user_id": user_id,
+        })
+
+        while not await request.is_disconnected():
+            try:
+                rows = _fetch_recent_logs(
+                    limit=max(1, min(limit, 100)),
+                    after_created_at=last_created_at,
+                    task_ids=task_ids,
+                )
+                for row in rows:
+                    row_id = row.get("id")
+                    if row_id in sent_ids:
+                        continue
+                    sent_ids.add(row_id)
+                    if len(sent_ids) > 500:
+                        sent_ids = set(list(sent_ids)[-250:])
+                    last_created_at = row.get("created_at") or last_created_at
+                    yield _sse_event("log", row, row_id)
+            except Exception as exc:
+                logger.warning("Agent log SSE stream failed to fetch logs: %s", exc)
+                yield _sse_event("error", {"message": str(exc)})
+
+            yield ": keep-alive\n\n"
+            await asyncio.sleep(1)
+
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no",
+        },
+    )
+
+
 @router.post("/{task_id}/run")
-async def run_task(task_id: str, background_tasks: BackgroundTasks):
+async def run_task(task_id: str, background_tasks: BackgroundTasks, use_queue: bool | None = None):
     """
     Triggers the execution of a specific task.
     """
@@ -64,6 +298,7 @@ async def run_task(task_id: str, background_tasks: BackgroundTasks):
         raise HTTPException(status_code=404, detail="Task not found")
     
     task = task_res.data
+    _assert_task_project_is_mutable(task)
     
     # 2. Check if agent is assigned
     agent_id = task.get("assigned_agent_id")
@@ -76,9 +311,30 @@ async def run_task(task_id: str, background_tasks: BackgroundTasks):
         raise HTTPException(status_code=404, detail="Assigned agent not found")
     
     agent_data = agent_res.data
+
+    should_queue = use_queue if use_queue is not None else settings.TASK_EXECUTION_MODE == "queue"
+    if should_queue:
+        queued = await TaskQueueService.queue_task(task_id)
+        if not queued or not queued.data:
+            raise HTTPException(status_code=500, detail="Task could not be queued")
+        await audit_service.log_action(
+            user_id=task.get("project", {}).get("owner_id"),
+            action="task_queued",
+            agent_id=agent_id,
+            task_id=task_id,
+            metadata={"project_id": task.get("project_id"), "source": "task_run_endpoint"},
+        )
+        return {"message": "Task queued for worker execution", "task_id": task_id, "mode": "queue"}
     
     # 4. Update task status to in_progress
     supabase.table("tasks").update({"status": "in_progress"}).eq("id", task_id).execute()
+    await audit_service.log_action(
+        user_id=task.get("project", {}).get("owner_id"),
+        action="task_run_started",
+        agent_id=agent_id,
+        task_id=task_id,
+        metadata={"project_id": task.get("project_id"), "mode": "direct"},
+    )
     
     # 5. Run in background
     background_tasks.add_task(AgentRunnerService.execute_agent_logic, task, agent_data)
@@ -86,23 +342,43 @@ async def run_task(task_id: str, background_tasks: BackgroundTasks):
     return {"message": "Task execution started", "task_id": task_id}
 
 @router.post("/{task_id}/approve")
-async def approve_task(task_id: str):
+async def approve_task(task_id: str, background_tasks: BackgroundTasks):
     task_res = supabase.table("tasks").select("*").eq("id", task_id).single().execute()
     if not task_res.data:
         raise HTTPException(status_code=404, detail="Task not found")
+    _assert_task_project_is_mutable(task_res.data)
     _assert_task_quality(task_res.data)
     task = update_task_status(task_id, "done")
+    
+    # Index for Long-Term Memory
+    background_tasks.add_task(memory_service.index_task_output, task)
+    
+    await audit_service.log_action(
+        user_id=None,
+        action="task_approved",
+        agent_id=task.get("assigned_agent_id"),
+        task_id=task_id,
+        metadata={"project_id": task.get("project_id")},
+    )
     return {"message": "Task approved", "task": task}
 
 @router.post("/{task_id}/reject")
 async def reject_task(task_id: str):
     task = update_task_status(task_id, "todo")
+    await audit_service.log_action(
+        user_id=None,
+        action="task_rejected",
+        agent_id=task.get("assigned_agent_id"),
+        task_id=task_id,
+        metadata={"project_id": task.get("project_id")},
+    )
     return {"message": "Task rejected", "task": task}
 @router.post("/project/{project_id}/approve-all")
-async def approve_all_tasks(project_id: str):
+async def approve_all_tasks(project_id: str, background_tasks: BackgroundTasks):
     """
     Approves all tasks in a project that are awaiting approval.
     """
+    _assert_project_is_mutable(project_id)
     waiting_tasks = (
         supabase.table("tasks")
         .select("*")
@@ -132,6 +408,10 @@ async def approve_all_tasks(project_id: str):
             .execute()
         )
         result_data = result.data or []
+        
+        # Index all approved tasks for Long-Term Memory
+        for approved_task in result_data:
+            background_tasks.add_task(memory_service.index_task_output, approved_task)
     
     # 2. Check if all tasks in project are now done
     task_result = (
@@ -143,6 +423,16 @@ async def approve_all_tasks(project_id: str):
     tasks = task_result.data or []
     if tasks and all(task.get("status") == "done" for task in tasks):
         supabase.table("projects").update({"status": "completed"}).eq("id", project_id).execute()
+
+    await audit_service.log_action(
+        user_id=None,
+        action="tasks_approved_bulk",
+        metadata={
+            "project_id": project_id,
+            "approved_count": len(result_data),
+            "blocked_count": len(blocked),
+        },
+    )
     
     return {
         "message": f"Approved {len(result_data)} tasks",

backend/services/agent_runner_service.py

@@ -2,12 +2,25 @@ import logging
 from datetime import datetime, timezone
 from services.supabase_service import supabase
 from services.audit_service import audit_service
+from services.budget_service import BudgetExceededError, budget_service
+from services.evidence_service import evidence_service
 from agents.agent_factory import AgentFactory
 from services.semantic_backprop import semantic_backprop
 from services.output_quality import build_quality_instructions, validate_output
+from services.memory_service import memory_service
 
 logger = logging.getLogger("agent_runner_service")
 
+def _update_task_run(run_id: str, payload: dict):
+    try:
+        return supabase.table("task_runs").update(payload).eq("id", run_id).execute()
+    except Exception as exc:
+        if "duration_seconds" in payload and "duration_seconds" in str(exc) and "schema cache" in str(exc):
+            fallback_payload = {key: value for key, value in payload.items() if key != "duration_seconds"}
+            logger.warning("task_runs.duration_seconds is missing in Supabase schema; retrying run update without duration.")
+            return supabase.table("task_runs").update(fallback_payload).eq("id", run_id).execute()
+        raise
+
 class AgentRunnerService:
     @staticmethod
     async def run_agent_task(
@@ -27,6 +40,13 @@ class AgentRunnerService:
 
         if update_task:
             supabase.table("tasks").update({"status": "in_progress"}).eq("id", task_id).execute()
+            await audit_service.log_action(
+                user_id=None,
+                action="task_status_changed",
+                agent_id=agent_data.get("id"),
+                task_id=task_id,
+                metadata={"project_id": project_id, "status": "in_progress"},
+            )
 
         try:
             run_res = supabase.table("task_runs").insert({
@@ -35,6 +55,13 @@ class AgentRunnerService:
                 "status": "running"
             }).execute()
             run_id = run_res.data[0]["id"]
+            await audit_service.log_action(
+                user_id=None,
+                action="task_run_created",
+                agent_id=agent_data.get("id"),
+                task_id=task_id,
+                metadata={"project_id": project_id, "run_id": run_id, "status": "running"},
+            )
 
             agent = AgentFactory.get_agent(
                 provider=agent_data["api_provider"],
@@ -65,6 +92,18 @@ class AgentRunnerService:
             extra_context = ""
             if include_semantic_context:
                 extra_context = await semantic_backprop.get_project_context(project_id, task_id)
+                # Fetch Long-Term Memory (Cross-project)
+                memories = await memory_service.search_memory(
+                    query=task.get("description") or task["title"],
+                    limit=3,
+                    threshold=0.72
+                )
+                if memories:
+                    memory_header = "\n\n### RELEVANT HISTORICAL CONTEXT (CROSS-PROJECT)\n"
+                    memory_blocks = []
+                    for m in memories:
+                        memory_blocks.append(f"- Memory: {m['content']}")
+                    extra_context += memory_header + "\n".join(memory_blocks)
 
             import time
             import hashlib
@@ -81,6 +120,7 @@ class AgentRunnerService:
             if cache_key in AgentRunnerService._task_cache:
                 logger.info(f"Cache hit for task {task_id}. Skipping LLM call.")
                 cached_result = AgentRunnerService._task_cache[cache_key]
+                claims_count = await evidence_service.replace_task_claims(task, cached_result)
                 
                 # Still log the "start" for UI consistency
                 agent_name = agent_data.get('name', 'Agent')
@@ -97,6 +137,24 @@ class AgentRunnerService:
                         "status": "awaiting_approval",
                         "output_data": cached_result
                     }).eq("id", task_id).execute()
+                    await audit_service.log_action(
+                        user_id=None,
+                        action="task_status_changed",
+                        agent_id=agent_data.get("id"),
+                        task_id=task_id,
+                        metadata={
+                            "project_id": project_id,
+                            "run_id": run_id,
+                            "status": "awaiting_approval",
+                            "cache_hit": True,
+                            "claims_count": claims_count,
+                        },
+                    )
+
+                _update_task_run(run_id, {
+                    "status": "completed",
+                    "finished_at": datetime.now(timezone.utc).isoformat()
+                })
                 
                 return cached_result, run_id
 
@@ -112,6 +170,24 @@ class AgentRunnerService:
             start_time = time.time()
             task_instructions = task.get("description") or task["title"]
             task_instructions = f"{task_instructions}\n\n{build_quality_instructions(quality_task)}"
+            prompt_tokens = budget_service.estimate_prompt_tokens(
+                task_instructions=task_instructions,
+                context=context,
+                extra_context=extra_context,
+                system_prompt=agent_data.get("system_prompt"),
+            )
+            max_completion_tokens = int(getattr(agent, "max_tokens", 0) or 0)
+            estimated_preflight_cost = budget_service.estimate_cost(
+                agent_data.get("api_provider"),
+                agent_data.get("model"),
+                prompt_tokens,
+                max_completion_tokens,
+            )
+            budget_service.check_before_run(
+                project_id=project_id,
+                estimated_tokens=prompt_tokens + max_completion_tokens,
+                estimated_cost=estimated_preflight_cost,
+            )
             result = await agent.run(task_instructions, context, extra_context=extra_context)
             duration = time.time() - start_time
 
@@ -129,6 +205,26 @@ class AgentRunnerService:
 
             quality_review = validate_output(quality_task, result)
             result["quality_review"] = quality_review
+            claims_count = await evidence_service.replace_task_claims(task, result)
+            completion_tokens = budget_service.estimate_completion_tokens(result)
+            estimated_cost = budget_service.estimate_cost(
+                agent_data.get("api_provider"),
+                agent_data.get("model"),
+                prompt_tokens,
+                completion_tokens,
+            )
+            budget_service.record_usage(
+                project_id=project_id,
+                task_id=task_id,
+                run_id=run_id,
+                agent_id=agent_data.get("id"),
+                provider=agent_data.get("api_provider"),
+                model=agent_data.get("model"),
+                prompt_tokens=prompt_tokens,
+                completion_tokens=completion_tokens,
+                estimated_cost=estimated_cost,
+                metadata={"duration_seconds": round(duration, 2), "claims_count": claims_count},
+            )
 
             # 6. Save to Cache
             AgentRunnerService._task_cache[cache_key] = result
@@ -138,13 +234,28 @@ class AgentRunnerService:
                     "status": "awaiting_approval",
                     "output_data": result
                 }).eq("id", task_id).execute()
+                await audit_service.log_action(
+                    user_id=None,
+                    action="task_status_changed",
+                    agent_id=agent_data.get("id"),
+                    task_id=task_id,
+                    metadata={
+                        "project_id": project_id,
+                        "run_id": run_id,
+                        "status": "awaiting_approval",
+                        "quality_approved": quality_review["approved"],
+                        "claims_count": claims_count,
+                        "estimated_tokens": prompt_tokens + completion_tokens,
+                        "estimated_cost": float(estimated_cost),
+                    },
+                )
 
             # 7. Update Run Status
-            supabase.table("task_runs").update({
+            _update_task_run(run_id, {
                 "status": "completed",
                 "finished_at": datetime.now(timezone.utc).isoformat(),
                 "duration_seconds": round(duration, 2)
-            }).eq("id", run_id).execute()
+            })
 
             # 8. Log Completion with Metrics
             supabase.table("agent_logs").insert({
@@ -161,22 +272,76 @@ class AgentRunnerService:
                     "action": "quality_review_failed",
                     "content": f"Quality review failed: {', '.join(quality_review['fail_reasons'])}"
                 }).execute()
+                await audit_service.log_action(
+                    user_id=None,
+                    action="task_quality_review_failed",
+                    agent_id=agent_data.get("id"),
+                    task_id=task_id,
+                    metadata={
+                        "project_id": project_id,
+                        "run_id": run_id,
+                        "fail_reasons": quality_review.get("fail_reasons", []),
+                    },
+                )
 
             return result, run_id
 
+        except BudgetExceededError as e:
+            logger.warning(f"Budget blocked task {task_id}: {str(e)}")
+            if run_id:
+                _update_task_run(run_id, {
+                    "status": "cancelled",
+                    "error_message": str(e),
+                    "finished_at": datetime.now(timezone.utc).isoformat()
+                })
+
+            if update_task:
+                supabase.table("tasks").update({
+                    "status": "failed",
+                    "output_data": {"error": str(e), "budget_blocked": True}
+                }).eq("id", task_id).execute()
+                await audit_service.log_action(
+                    user_id=None,
+                    action="task_budget_blocked",
+                    agent_id=agent_data.get("id"),
+                    task_id=task_id,
+                    metadata={"project_id": project_id, "run_id": run_id, "error": str(e)},
+                )
+
+            supabase.table("agent_logs").insert({
+                "task_id": task_id,
+                "run_id": run_id,
+                "action": "budget_blocked",
+                "content": f"Budget blocked execution: {str(e)}"
+            }).execute()
+
+            raise e
+
         except Exception as e:
             logger.error(f"Error executing task {task_id}: {str(e)}")
             if run_id:
-                supabase.table("task_runs").update({
+                _update_task_run(run_id, {
                     "status": "failed",
                     "finished_at": datetime.now(timezone.utc).isoformat()
-                }).eq("id", run_id).execute()
+                })
             
             if update_task:
                 supabase.table("tasks").update({
                     "status": "failed",
                     "output_data": {"error": str(e)}
                 }).eq("id", task_id).execute()
+                await audit_service.log_action(
+                    user_id=None,
+                    action="task_status_changed",
+                    agent_id=agent_data.get("id"),
+                    task_id=task_id,
+                    metadata={
+                        "project_id": project_id,
+                        "run_id": run_id,
+                        "status": "failed",
+                        "error": str(e),
+                    },
+                )
 
             # LOG ERROR TO AGENT CONSOLE
             supabase.table("agent_logs").insert({

backend/services/memory_service.py

@@ -0,0 +1,112 @@
+import logging
+from typing import List, Dict, Any, Optional
+from services.supabase_service import supabase
+from services.embedding_service import embedding_service
+
+logger = logging.getLogger("uvicorn")
+
+class MemoryService:
+    """
+    Handles vectorized long-term memory for Aubm projects.
+    Allows agents to retrieve context from past projects and approved work.
+    """
+
+    async def save_memory(
+        self,
+        project_id: str,
+        content: str,
+        task_id: Optional[str] = None,
+        memory_type: str = "approved_output",
+        metadata: Optional[Dict[str, Any]] = None
+    ) -> bool:
+        """
+        Vectorizes content and saves it to project_memory.
+        """
+        try:
+            if not content or len(content.strip()) < 10:
+                return False
+
+            embedding = await embedding_service.get_embedding(content)
+            
+            data = {
+                "project_id": project_id,
+                "task_id": task_id,
+                "content": content,
+                "embedding": embedding,
+                "memory_type": memory_type,
+                "metadata": metadata or {}
+            }
+            
+            result = supabase.table("project_memory").insert(data).execute()
+            return len(result.data) > 0
+        except Exception as e:
+            logger.error(f"Failed to save memory: {e}")
+            return False
+
+    async def search_memory(
+        self,
+        query: str,
+        limit: int = 5,
+        threshold: float = 0.7,
+        project_id: Optional[str] = None
+    ) -> List[Dict[str, Any]]:
+        """
+        Performs semantic search across project memory.
+        If project_id is provided, filters memory to that project only (short-term).
+        If project_id is None, searches cross-project (long-term).
+        """
+        try:
+            query_embedding = await embedding_service.get_embedding(query)
+            
+            # Use the match_project_memory RPC function defined in add_vector_memory.sql
+            rpc_params = {
+                "query_embedding": query_embedding,
+                "match_threshold": threshold,
+                "match_count": limit,
+            }
+            
+            if project_id:
+                rpc_params["filter_project_id"] = project_id
+
+            result = supabase.rpc("match_project_memory", rpc_params).execute()
+            return result.data or []
+        except Exception as e:
+            logger.error(f"Failed to search memory: {e}")
+            return []
+
+    async def index_task_output(self, task: Dict[str, Any]) -> bool:
+        """
+        Specialized indexer for approved task outputs.
+        """
+        output_data = task.get("output_data")
+        if not output_data:
+            return False
+            
+        # Extract meaningful text from output
+        content = ""
+        if isinstance(output_data, str):
+            content = output_data
+        elif isinstance(output_data, dict):
+            # Try to get the primary content
+            content = (
+                output_data.get("data") or 
+                output_data.get("strategicConclusion") or 
+                output_data.get("raw_output") or
+                str(output_data)
+            )
+        
+        if not content:
+            return False
+
+        return await self.save_memory(
+            project_id=task.get("project_id"),
+            task_id=task.get("id"),
+            content=str(content),
+            memory_type="approved_output",
+            metadata={
+                "task_title": task.get("title"),
+                "agent_id": task.get("assigned_agent_id")
+            }
+        )
+
+memory_service = MemoryService()