fix/editor flow

apps/api/app/api/routes/agent.py

@@ -10,11 +10,13 @@
 import asyncio
 import json
 import logging
+import re
 import uuid
 from collections.abc import AsyncGenerator
+from dataclasses import dataclass
 from typing import Any, Literal
 
-from fastapi import APIRouter, Depends, Request
+from fastapi import APIRouter, Depends
 from fastapi.responses import StreamingResponse
 from openai import AsyncOpenAI
 from pydantic import BaseModel
@@ -34,6 +36,80 @@
 log = logging.getLogger("fiebatt.agent")
 router = APIRouter(prefix="/agent", tags=["agent"])
 
+
+_TOOL_MARKUP_START = re.compile(
+    r"<\s*(?:(?:[|｜]\s*)?DSML(?:\s*[|｜])?\s*)?"
+    r"(?:function[\s_]?calls?|tool[\s_]?calls?|invoke\b|parameter\b)",
+    re.IGNORECASE,
+)
+_DSML_INVOKE = re.compile(
+    r'<\s*(?:(?:[|｜]\s*)?DSML(?:\s*[|｜])?\s*)?invoke\s+name="([^"]+)"[^>]*>'
+    r"(.*?)"
+    r"</\s*(?:(?:[|｜]\s*)?DSML(?:\s*[|｜])?\s*)?invoke\s*>",
+    re.IGNORECASE | re.DOTALL,
+)
+_DSML_PARAMETER = re.compile(
+    r'<\s*(?:(?:[|｜]\s*)?DSML(?:\s*[|｜])?\s*)?parameter\s+'
+    r'name="([^"]+)"(?:\s+string="(true|false)")?[^>]*>'
+    r"(.*?)"
+    r"</\s*(?:(?:[|｜]\s*)?DSML(?:\s*[|｜])?\s*)?parameter\s*>",
+    re.IGNORECASE | re.DOTALL,
+)
+
+
+@dataclass(slots=True)
+class _ParsedFunction:
+    name: str
+    arguments: str
+
+
+@dataclass(slots=True)
+class _ParsedToolCall:
+    id: str
+    function: _ParsedFunction
+
+
+def _clean_agent_text(text: str) -> str:
+    """Return user-facing prose before any leaked tool protocol block."""
+    marker = _TOOL_MARKUP_START.search(text)
+    visible = text[:marker.start()] if marker else text
+    return visible.strip()
+
+
+def _parse_dsml_tool_calls(text: str) -> list[_ParsedToolCall]:
+    """Recover DeepSeek-style DSML calls leaked into message content."""
+    parsed: list[_ParsedToolCall] = []
+    for index, invoke in enumerate(_DSML_INVOKE.finditer(text)):
+        name, body = invoke.groups()
+        arguments: dict[str, Any] = {}
+        for parameter in _DSML_PARAMETER.finditer(body):
+            key, is_string, raw_value = parameter.groups()
+            value = raw_value.strip()
+            if (is_string or "false").lower() != "true":
+                try:
+                    arguments[key] = json.loads(value)
+                except json.JSONDecodeError:
+                    arguments[key] = value
+            else:
+                arguments[key] = value
+        if not arguments:
+            try:
+                direct = json.loads(body.strip())
+                if isinstance(direct, dict):
+                    arguments = direct
+            except json.JSONDecodeError:
+                pass
+        parsed.append(
+            _ParsedToolCall(
+                id=f"dsml-{uuid.uuid4().hex[:12]}-{index}",
+                function=_ParsedFunction(
+                    name=name,
+                    arguments=json.dumps(arguments),
+                ),
+            )
+        )
+    return parsed
+
 # ---- request schema ----
 
 class BBoxParam(BaseModel):
@@ -761,9 +837,9 @@ async def _bridge_plan_events(
 
     The ``generate_job`` worker publishes a rich trail of stage events on
     the per-job event bus (plan_start, plan_done, gen_start, variant_done…).
-    The Pro-mode reveal panel consumes these directly via
-    ``/api/jobs/{id}/stream``, but the Vibe-mode agent chat only hears the
-    agent SSE stream — so by default the user never sees that the model is
+    The detailed generation panel consumes these directly via
+    ``/api/jobs/{id}/stream``, while agent chat only hears the agent SSE
+    stream — so by default the user never sees that the model is
     quietly rewriting their one-line prompt into a video generation brief.
 
     This helper bridges the gap for exactly the events that are interesting
@@ -887,7 +963,6 @@ def _build_messages(
 
 async def _agent_stream(
     *,
-    request: Request,
     body: AgentChatRequest,
     session_id: str,
     runner: Any,
@@ -988,18 +1063,17 @@ async def _agent_stream(
             )
             choice = response.choices[0]
             msg = choice.message
-            text = msg.content or ""
-            tool_calls = msg.tool_calls or []
+            raw_text = msg.content or ""
+            tool_calls: list[Any] = list(msg.tool_calls or [])
+            if not tool_calls:
+                tool_calls = _parse_dsml_tool_calls(raw_text)
+            text = _clean_agent_text(raw_text)
         except Exception as exc:
             log.exception("[agent.chat] req=%s %s API call failed on turn %d", req_id, provider_name, turn)
             yield sse_event("error", {"message": f"{provider_name} API error: {exc}"})
             yield sse_event("done", {})
             return
 
-        if await request.is_disconnected():
-            log.info("Client disconnected during agent loop")
-            return
-
         if text:
             _agent_text_parts.append(text)
             yield sse_event("token", {"text": text})
@@ -1036,6 +1110,10 @@ async def _agent_stream(
             tool_args = json.loads(tc.function.arguments) if tc.function.arguments else {}
             if tool_name == "create_edit_plan" and body.selection_id:
                 tool_args.setdefault("selection_id", body.selection_id)
+            if tool_name == "generate_edit":
+                # Keep attribution separate from the model-authored generation
+                # prompt. The worker emits both for the before/refined card.
+                tool_args["user_prompt"] = body.message
             if tool_name in {"create_edit_plan", "generate_edit"} and body.video_gen_provider:
                 tool_args["video_gen_provider"] = body.video_gen_provider
 
@@ -1186,13 +1264,47 @@ async def _agent_stream(
     yield sse_event("done", {})
 
 
+# Keep strong references so Python cannot collect detached agent turns while
+# Qwen is planning or queueing their generation job.
+_agent_turn_tasks: set[asyncio.Task[None]] = set()
+
+
+async def _detached_agent_relay(
+    source: AsyncGenerator[str, None],
+) -> AsyncGenerator[str, None]:
+    """Relay SSE while letting agent orchestration survive client navigation."""
+    queue: asyncio.Queue[str | None] = asyncio.Queue()
+
+    async def consume() -> None:
+        try:
+            async for event in source:
+                await queue.put(event)
+        except Exception as exc:
+            log.exception("detached agent turn failed")
+            await queue.put(sse_event("error", {"message": str(exc)}))
+            await queue.put(sse_event("done", {}))
+        finally:
+            await queue.put(None)
+
+    task = asyncio.create_task(consume(), name=f"agent-turn:{uuid.uuid4().hex[:8]}")
+    _agent_turn_tasks.add(task)
+    task.add_done_callback(_agent_turn_tasks.discard)
+
+    # Closing this relay only drops browser delivery. The consumer keeps
+    # running, persists its conversation, and starts any requested video job.
+    while True:
+        event = await queue.get()
+        if event is None:
+            return
+        yield event
+
+
 # ---- route ----
 
 
 @router.post("/chat")
 async def agent_chat(
     body: AgentChatRequest,
-    request: Request,
     session: SessionModel = Depends(get_session),
     runner=Depends(get_runner),
 ) -> StreamingResponse:
@@ -1205,11 +1317,12 @@ async def agent_chat(
     needs them instead.
     """
     return StreamingResponse(
-        _agent_stream(
-            request=request,
-            body=body,
-            session_id=session.id,
-            runner=runner,
+        _detached_agent_relay(
+            _agent_stream(
+                body=body,
+                session_id=session.id,
+                runner=runner,
+            ),
         ),
         media_type="text/event-stream",
         headers={

apps/api/app/api/routes/jobs.py

@@ -1,7 +1,7 @@
 import asyncio
 import json
 
-from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
 from fastapi.responses import StreamingResponse
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -18,31 +18,15 @@
 router = APIRouter(tags=["jobs"])
 
 
-@router.get("/jobs/{job_id}", response_model=JobOut)
-async def get_job(
-    job_id: str,
-    session: SessionModel = Depends(get_session),
-    db: AsyncSession = Depends(get_db),
-):
-    job = (
-        await db.execute(
-            select(Job).where(Job.id == job_id).options(selectinload(Job.variants))
-        )
-    ).scalar_one_or_none()
-    if job is None:
-        raise HTTPException(status_code=404, detail="job not found")
-
-    # enforce session ownership through the project
-    proj = await db.get(Project, job.project_id)
-    if proj is None or proj.session_id != session.id:
-        raise HTTPException(status_code=404, detail="job not found")
-
+def _job_out(job: Job) -> JobOut:
     payload = job.payload or {}
     return JobOut(
         job_id=job.id,
         kind=job.kind,
         status=job.status,  # type: ignore[arg-type]
         error=job.error,
+        created_at=job.created_at,
+        accepted=bool(payload.get("latest_accepted_segment_id")),
         start_ts=job.start_ts,
         end_ts=job.end_ts,
         provider=payload.get("selected_provider"),
@@ -74,6 +58,51 @@ async def get_job(
     )
 
 
+@router.get("/projects/{project_id}/generation-jobs", response_model=list[JobOut])
+async def list_generation_jobs(
+    project_id: str,
+    limit: int = Query(default=10, ge=1, le=20),
+    session: SessionModel = Depends(get_session),
+    db: AsyncSession = Depends(get_db),
+):
+    """Recent generation jobs used to restore work after page navigation."""
+    proj = await db.get(Project, project_id)
+    if proj is None or proj.session_id != session.id:
+        raise HTTPException(status_code=404, detail="project not found")
+    rows = (
+        await db.execute(
+            select(Job)
+            .where(Job.project_id == project_id, Job.kind == "generate")
+            .options(selectinload(Job.variants))
+            .order_by(Job.created_at.desc())
+            .limit(limit)
+        )
+    ).scalars().all()
+    return [_job_out(job) for job in rows]
+
+
+@router.get("/jobs/{job_id}", response_model=JobOut)
+async def get_job(
+    job_id: str,
+    session: SessionModel = Depends(get_session),
+    db: AsyncSession = Depends(get_db),
+):
+    job = (
+        await db.execute(
+            select(Job).where(Job.id == job_id).options(selectinload(Job.variants))
+        )
+    ).scalar_one_or_none()
+    if job is None:
+        raise HTTPException(status_code=404, detail="job not found")
+
+    # enforce session ownership through the project
+    proj = await db.get(Project, job.project_id)
+    if proj is None or proj.session_id != session.id:
+        raise HTTPException(status_code=404, detail="job not found")
+
+    return _job_out(job)
+
+
 @router.get("/jobs/{job_id}/stream")
 async def stream_job_events(
     job_id: str,

apps/api/app/schemas/job.py

@@ -1,3 +1,4 @@
+from datetime import datetime
 from typing import Any
 
 from pydantic import BaseModel, Field
@@ -21,6 +22,8 @@ class JobOut(BaseModel):
     kind: str
     status: JobStatus
     error: str | None = None
+    created_at: datetime | None = None
+    accepted: bool = False
     variants: list[VariantOut] = []
     # authoritative edit window used when accepting a generated replacement.
     start_ts: float | None = None

apps/api/app/services/agent_tools.py

@@ -655,6 +655,7 @@ async def _generate_edit(
         reference_frame_ts=reference_frame_ts,
         payload={
             "video_gen_provider": video_gen_provider or "auto",
+            "user_prompt": str(args.get("user_prompt") or prompt),
             "plan_id": plan.id if plan else None,
             "planned_context": plan.range_json if plan else None,
             "planned_intent": plan.intent_json if plan else None,

apps/api/app/workers/generate_job.py

@@ -588,7 +588,7 @@ async def _do_extract_end_frame() -> tuple[Path, bool]:
             if planned_intent is not None and planned_intent.grounded_edit is not None
             else "asking Qwen to structure the edit plan"
         ),
-        user_prompt=prompt,
+        user_prompt=str(payload.get("user_prompt") or prompt),
         bbox=bbox,
         duration=segment_duration,
     )