Add self-sent knowledge WebDAV intent

AGENTS.md

@@ -108,6 +108,16 @@
   stay idempotent per source email, preserve email/thread provenance, and store
   only plain-text task titles. Do not create unlinked knowledge tasks from raw
   dict payloads when the source email row is unavailable.
+- Self-sent knowledge WebDAV/Notes materialization requests must start from an
+  opaque task uid, re-check owner and organization scope server-side, reject
+  non-`self_sent_knowledge` tasks, and return intent metadata only until the
+  connector/provider write path has source-backed ETag conflict handling.
+- Private API services should return deterministic `error_code` values for
+  expected failures; route layers must not derive HTTP status from human-readable
+  message substrings.
+- UI async state for repeated task/action rows must be keyed by the row's opaque
+  public id so one row's loading, error, or result state cannot overwrite another
+  row.
 - Sender ontology and relationship DAG APIs must stay scoped to the signed
   session owner and organization. Source-backed UI panels must request
   `source_message_id` and `source_thread_id` filters instead of presenting a

README.md

@@ -42,8 +42,10 @@ mail/calendar/file systems.
   still an episode task.
 - **Self-sent knowledge capture**: IMAP-imported emails sent from a user to the
   same address now create one idempotent, source-linked `self_sent_knowledge`
-  ticket task with a plain-text memo title. WebDAV/Notes materialization remains
-  future work and must write back to customer-owned sources.
+  ticket task with a plain-text memo title. The Tasks workspace can request a
+  signed WebDAV/Notes materialization intent for that task and shows the planned
+  customer-owned target with `provider_write_executed=false`; actual provider
+  mutation remains connector execution work.
 
 ## Five-minute local path
 

backend/api/webdav.py

@@ -1,4 +1,4 @@
-from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel
 from sqlalchemy.ext.asyncio import AsyncSession
 from typing import List
@@ -8,6 +8,13 @@
 from services.webdav_service import webdav_service
 
 router = APIRouter(prefix="/api/webdav", tags=["webdav"])
+WEB_DAV_ERROR_STATUS_CODES = {
+    "not_found": 404,
+    "validation_error": 422,
+    "missing_provenance": 422,
+    "no_webdav_account": 422,
+    "webdav_account_not_found": 422,
+}
 
 class WebdavAccountResponse(BaseModel):
     account_id: int
@@ -31,6 +38,25 @@ class WritebackIntentResponse(BaseModel):
     status: str | None = None
     message: str | None = None
 
+class KnowledgeMaterializationIntentRequest(BaseModel):
+    source_task_id: str
+    target_account_id: int | None = None
+
+class KnowledgeMaterializationIntentResponse(BaseModel):
+    intent: str
+    status: str
+    task_id: str
+    source_type: str
+    source_email_id: str | None
+    source_thread_id: str | None
+    source_id: int | None
+    server_url: str | None
+    target_path: str
+    requires_if_match: bool
+    provenance: str
+    provider_write_executed: bool
+    audit_event: str
+
 @router.get("/accounts", response_model=List[WebdavAccountResponse])
 async def get_webdav_accounts(
     auth_context: AuthContext = Depends(get_auth_context),
@@ -62,3 +88,27 @@ async def get_webdav_writeback_intent(
     if result.get("status") == "error":
         raise HTTPException(status_code=422, detail=result.get("message"))
     return WritebackIntentResponse(**result)
+
+@router.post(
+    "/knowledge-materialization-intent",
+    response_model=KnowledgeMaterializationIntentResponse,
+)
+async def get_knowledge_materialization_intent(
+    req: KnowledgeMaterializationIntentRequest,
+    auth_context: AuthContext = Depends(get_auth_context),
+    db: AsyncSession = Depends(get_db),
+):
+    result = await webdav_service.determine_knowledge_materialization_intent_from_db(
+        db,
+        auth_context.user_id,
+        auth_context.organization_id,
+        req.source_task_id,
+        target_account_id=req.target_account_id,
+    )
+    if result.get("status") == "error":
+        status_code = WEB_DAV_ERROR_STATUS_CODES.get(
+            str(result.get("error_code") or ""),
+            422,
+        )
+        raise HTTPException(status_code=status_code, detail=result.get("message"))
+    return KnowledgeMaterializationIntentResponse(**result)

backend/services/webdav_service.py

@@ -4,7 +4,8 @@
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from db.models import ProjectFolder, WebdavAccount
+from db.models import Email, ProjectFolder, TicketTask, WebdavAccount
+from services.knowledge_extractor import SELF_SENT_KNOWLEDGE_SOURCE
 
 logger = logging.getLogger(__name__)
 
@@ -124,13 +125,85 @@ async def determine_webdav_writeback_intent_from_db(
             target_account_id=target_account_id,
         )
 
+    async def determine_knowledge_materialization_intent_from_db(
+        self,
+        session: AsyncSession,
+        user_id: str,
+        organization_id: str | None,
+        source_task_id: str,
+        target_account_id: int | None = None,
+    ) -> Dict[str, Any]:
+        task_result = await session.execute(
+            select(TicketTask, Email.message_id)
+            .outerjoin(
+                Email,
+                (TicketTask.related_email_id == Email.id)
+                & (Email.user_id == user_id)
+                & (Email.organization_id == organization_id),
+            )
+            .where(
+                TicketTask.task_uid == source_task_id,
+                TicketTask.user_id == user_id,
+                TicketTask.organization_id == organization_id,
+            )
+        )
+        row = task_result.one_or_none()
+        if row is None:
+            return {
+                "status": "error",
+                "error_code": "not_found",
+                "message": "Self-sent knowledge task was not found.",
+            }
+
+        task, source_email_id = row
+        if task.source_type != SELF_SENT_KNOWLEDGE_SOURCE:
+            return {
+                "status": "error",
+                "error_code": "validation_error",
+                "message": "Task is not self-sent knowledge.",
+            }
+        if source_email_id is None:
+            return {
+                "status": "error",
+                "error_code": "missing_provenance",
+                "message": "Self-sent knowledge task missing source email provenance.",
+            }
+
+        result = await self.determine_webdav_writeback_intent_from_db(
+            session,
+            user_id,
+            target_account_id=target_account_id,
+        )
+        if result.get("status") == "error":
+            return result
+
+        return {
+            "intent": "knowledge_materialization",
+            "status": "intent_ready",
+            "task_id": task.task_uid,
+            "source_type": SELF_SENT_KNOWLEDGE_SOURCE,
+            "source_email_id": source_email_id,
+            "source_thread_id": task.related_thread_id,
+            "source_id": result["source_id"],
+            "server_url": result["server_url"],
+            "target_path": f"/Naruon/Notes/{task.task_uid}.md",
+            "requires_if_match": result["requires_if_match"],
+            "provenance": result["provenance"],
+            "provider_write_executed": False,
+            "audit_event": "webdav.self_sent_knowledge_intent.created",
+        }
+
     def determine_webdav_writeback_intent_from_accounts(
         self,
         accounts: List[Dict[str, Any]],
         target_account_id: int | None = None,
     ) -> Dict[str, Any]:
         if not accounts:
-            return {"status": "error", "message": "No connected WebDAV accounts found."}
+            return {
+                "status": "error",
+                "error_code": "no_webdav_account",
+                "message": "No connected WebDAV accounts found.",
+            }
 
         selected_account = accounts[0]
         if target_account_id is not None:
@@ -142,6 +215,7 @@ def determine_webdav_writeback_intent_from_accounts(
             if selected_account is None:
                 return {
                     "status": "error",
+                    "error_code": "webdav_account_not_found",
                     "message": "Requested WebDAV account was not found.",
                 }