157 lines
6.5 KiB
Python
157 lines
6.5 KiB
Python
"""Conversation and attachment storage for the CAD delivery boundary.
|
|
|
|
CAD task state deliberately does not live here. The Authoring protocol owns mutable task
|
|
state in ``SqliteTaskRepository`` and immutable task artifacts in
|
|
``FileArtifactStore``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
import re
|
|
import secrets
|
|
from typing import Any
|
|
|
|
from app.settings import Settings
|
|
|
|
|
|
TASK_ID = re.compile(r"^cad_[a-z0-9]{12}$")
|
|
CONVERSATION_ID = re.compile(r"^conv_[a-z0-9]{12}$")
|
|
|
|
|
|
def now_iso() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def safe_task_id(task_id: str) -> str:
|
|
value = str(task_id or "").strip()
|
|
if not TASK_ID.fullmatch(value):
|
|
raise ValueError("Invalid task id")
|
|
return value
|
|
|
|
|
|
def safe_conversation_id(conversation_id: str) -> str:
|
|
value = str(conversation_id or "").strip()
|
|
if not CONVERSATION_ID.fullmatch(value):
|
|
raise ValueError("Invalid conversation id")
|
|
return value
|
|
|
|
|
|
def _safe_relative_path(value: str) -> str:
|
|
path = Path(str(value or ""))
|
|
if not value or path.is_absolute() or ".." in path.parts:
|
|
raise ValueError("Invalid conversation artifact path")
|
|
return path.as_posix()
|
|
|
|
|
|
def _write_json(path: Path, payload: Any) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
|
|
|
|
def _read_json(path: Path) -> dict[str, Any] | None:
|
|
if not path.is_file():
|
|
return None
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
return value if isinstance(value, dict) else None
|
|
|
|
|
|
class WorkspaceStore:
|
|
"""A conversation store, intentionally not a CAD task repository."""
|
|
|
|
def __init__(self, settings: Settings) -> None:
|
|
self.settings = settings
|
|
self.settings.conversation_root.mkdir(parents=True, exist_ok=True)
|
|
|
|
def conversation_dir(self, conversation_id: str) -> Path:
|
|
return self.settings.conversation_root / safe_conversation_id(conversation_id)
|
|
|
|
def conversation_path(self, conversation_id: str) -> Path:
|
|
return self.conversation_dir(conversation_id) / "conversation.json"
|
|
|
|
def ensure_conversation(self, conversation_id: str | None, current_task_id: str | None = None) -> dict[str, Any]:
|
|
conversation = safe_conversation_id(conversation_id) if conversation_id else f"conv_{secrets.token_hex(6)}"
|
|
path = self.conversation_path(conversation)
|
|
current = _read_json(path)
|
|
if current is not None:
|
|
if current_task_id and current.get("current_task_id") != safe_task_id(current_task_id):
|
|
current["current_task_id"] = safe_task_id(current_task_id)
|
|
current["updated_at"] = now_iso()
|
|
_write_json(path, current)
|
|
return current
|
|
record = {
|
|
"schema_version": "3.0",
|
|
"conversation_id": conversation,
|
|
"created_at": now_iso(),
|
|
"updated_at": now_iso(),
|
|
"current_task_id": safe_task_id(current_task_id) if current_task_id else "",
|
|
"messages": [],
|
|
"attachments": [],
|
|
}
|
|
_write_json(path, record)
|
|
return record
|
|
|
|
def read_conversation(self, conversation_id: str) -> dict[str, Any] | None:
|
|
return _read_json(self.conversation_path(conversation_id))
|
|
|
|
def clear_current_task_references(self) -> int:
|
|
"""Detach conversations from task artifacts removed by a protocol reset."""
|
|
cleared = 0
|
|
for path in self.settings.conversation_root.glob("conv_*/conversation.json"):
|
|
record = _read_json(path)
|
|
if not isinstance(record, dict) or not record.get("current_task_id"):
|
|
continue
|
|
record["current_task_id"] = ""
|
|
record["updated_at"] = now_iso()
|
|
_write_json(path, record)
|
|
cleared += 1
|
|
return cleared
|
|
|
|
def append_conversation_message(self, conversation_id: str, message: dict[str, Any], current_task_id: str | None = None) -> dict[str, Any]:
|
|
record = self.ensure_conversation(conversation_id, current_task_id)
|
|
known = {str(item.get("id") or "") for item in record["messages"] if isinstance(item, dict)}
|
|
if str(message.get("id") or "") not in known:
|
|
record["messages"].append(message)
|
|
if current_task_id:
|
|
record["current_task_id"] = safe_task_id(current_task_id)
|
|
record["updated_at"] = now_iso()
|
|
_write_json(self.conversation_path(record["conversation_id"]), record)
|
|
return record
|
|
|
|
def conversation_attachment_path(self, conversation_id: str, relative_path: str) -> Path:
|
|
root = self.conversation_dir(conversation_id).resolve()
|
|
candidate = (root / _safe_relative_path(relative_path)).resolve()
|
|
if root != candidate and root not in candidate.parents:
|
|
raise ValueError("Attachment path escapes conversation directory")
|
|
return candidate
|
|
|
|
def write_conversation_upload(self, conversation_id: str, filename: str, data: bytes) -> tuple[str, Path]:
|
|
conversation = safe_conversation_id(conversation_id)
|
|
safe_name = re.sub(r"[^a-zA-Z0-9._-]+", "_", Path(filename).name).strip("._") or "attachment"
|
|
relative = (Path("uploads") / f"upload_{secrets.token_hex(6)}_{safe_name}").as_posix()
|
|
target = self.conversation_attachment_path(conversation, relative)
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(data)
|
|
return relative, target
|
|
|
|
def add_conversation_attachment(self, conversation_id: str, attachment: dict[str, Any]) -> dict[str, Any]:
|
|
conversation = safe_conversation_id(conversation_id)
|
|
record = self.read_conversation(conversation)
|
|
if record is None:
|
|
raise ValueError("Conversation not found")
|
|
if str(attachment.get("conversation_id") or "") != conversation:
|
|
raise ValueError("Attachment does not belong to this conversation")
|
|
attachment_id = str(attachment.get("id") or "")
|
|
if not attachment_id:
|
|
raise ValueError("Attachment id is required")
|
|
self.conversation_attachment_path(conversation, str(attachment.get("path") or ""))
|
|
attachments = record.setdefault("attachments", [])
|
|
if any(str(item.get("id") or "") == attachment_id for item in attachments if isinstance(item, dict)):
|
|
raise ValueError("Attachment already exists")
|
|
attachments.append(attachment)
|
|
record["updated_at"] = now_iso()
|
|
_write_json(self.conversation_path(conversation), record)
|
|
return record
|