feat(opm): add validation, extraction, schemas, and tests

web/app/db.py

@@ -46,6 +46,16 @@ def init_db() -> None:
             );
             """
         )
+        cursor.execute(
+            """
+            CREATE TABLE IF NOT EXISTS opm_diagrams (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                note_id INTEGER,
+                payload TEXT NOT NULL,
+                created_at TEXT DEFAULT (datetime('now'))
+            );
+            """
+        )
         connection.commit()
 
 
@@ -114,3 +124,52 @@ def mark_action_item_done(action_item_id: int, done: bool) -> None:
         connection.commit()
 
 
+import json as _json
+
+
+def insert_opm_diagram(payload: dict, note_id: Optional[int] = None) -> int:
+    with get_connection() as connection:
+        cursor = connection.cursor()
+        cursor.execute(
+            "INSERT INTO opm_diagrams (note_id, payload) VALUES (?, ?)",
+            (note_id, _json.dumps(payload)),
+        )
+        connection.commit()
+        return int(cursor.lastrowid)
+
+
+def list_opm_diagrams() -> list[dict]:
+    with get_connection() as connection:
+        cursor = connection.cursor()
+        cursor.execute(
+            "SELECT id, note_id, payload, created_at FROM opm_diagrams ORDER BY id DESC"
+        )
+        rows = cursor.fetchall()
+    return [
+        {
+            "id": r["id"],
+            "note_id": r["note_id"],
+            "created_at": r["created_at"],
+            "diagram": _json.loads(r["payload"]),
+        }
+        for r in rows
+    ]
+
+
+def get_opm_diagram(diagram_id: int) -> Optional[dict]:
+    with get_connection() as connection:
+        cursor = connection.cursor()
+        cursor.execute(
+            "SELECT id, note_id, payload, created_at FROM opm_diagrams WHERE id = ?",
+            (diagram_id,),
+        )
+        row = cursor.fetchone()
+    if row is None:
+        return None
+    return {
+        "id": row["id"],
+        "note_id": row["note_id"],
+        "created_at": row["created_at"],
+        "diagram": _json.loads(row["payload"]),
+    }
+

web/app/main.py

@@ -8,7 +8,7 @@
 from fastapi.staticfiles import StaticFiles
 
 from .db import init_db
-from .routers import action_items, notes
+from .routers import action_items, notes, opm
 from . import db
 
 init_db()
@@ -24,6 +24,7 @@ def index() -> str:
 
 app.include_router(notes.router)
 app.include_router(action_items.router)
+app.include_router(opm.router)
 
 
 static_dir = Path(__file__).resolve().parents[1] / "frontend"

web/app/routers/opm.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from typing import Any, Dict, Optional
+
+from fastapi import APIRouter, HTTPException
+from pydantic import ValidationError
+
+from .. import db
+from ..services.opm_extract import extract_opm_diagram
+from ..services.opm_validate import validate_diagram
+
+
+router = APIRouter(prefix="/opm", tags=["opm"])
+
+
+@router.post("/extract")
+def extract(payload: Dict[str, Any]) -> Dict[str, Any]:
+    text = str(payload.get("text", "")).strip()
+    if not text:
+        raise HTTPException(status_code=400, detail="text is required")
+
+    note_id: Optional[int] = None
+    if payload.get("save_note"):
+        note_id = db.insert_note(text)
+
+    raw_dict = extract_opm_diagram(text)
+
+    try:
+        validated = validate_diagram(raw_dict)
+    except (ValidationError, ValueError) as exc:
+        raise HTTPException(
+            status_code=422,
+            detail={
+                "error": "opm_extraction_failed",
+                "stage": "validation",
+                "detail": str(exc),
+            },
+        )
+
+    diagram_id = db.insert_opm_diagram(validated.model_dump(), note_id=note_id)
+    return {"note_id": note_id, "diagram_id": diagram_id, "diagram": validated.model_dump()}
+
+
+@router.get("")
+def list_all() -> Dict[str, Any]:
+    return {"diagrams": db.list_opm_diagrams()}
+
+
+@router.get("/{diagram_id}")
+def get_one(diagram_id: int) -> Dict[str, Any]:
+    row = db.get_opm_diagram(diagram_id)
+    if row is None:
+        raise HTTPException(status_code=404, detail="diagram not found")
+    return row

web/app/schemas/opm.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from enum import Enum
+from typing import List
+
+from pydantic import BaseModel, Field, model_validator
+
+
+class NodeKind(str, Enum):
+    object = "object"
+    process = "process"
+    state = "state"
+
+
+class Relation(str, Enum):
+    agent = "agent"
+    instrument = "instrument"
+    consumption = "consumption"
+    result = "result"
+    effect = "effect"
+    aggregation = "aggregation"
+    specialization = "specialization"
+    characterization = "characterization"
+
+
+class OpmNode(BaseModel):
+    id: str = Field(..., pattern=r'^[a-z0-9]+(-[a-z0-9]+)*$')
+    kind: NodeKind
+    label: str = Field(..., min_length=1, max_length=80)
+
+
+class OpmLink(BaseModel):
+    id: str = Field(..., min_length=1)
+    source: str
+    target: str
+    relation: Relation
+
+
+class OpmDiagram(BaseModel):
+    version: str = Field(default="1.0")
+    nodes: List[OpmNode]
+    links: List[OpmLink]
+
+    @model_validator(mode="after")
+    def check_graph_integrity(self) -> "OpmDiagram":
+        node_ids = {n.id for n in self.nodes}
+
+        # Rule: node IDs must be unique
+        if len(node_ids) != len(self.nodes):
+            raise ValueError("Duplicate node IDs detected")
+
+        # Rule: link IDs must be unique
+        link_ids = [lk.id for lk in self.links]
+        if len(set(link_ids)) != len(link_ids):
+            raise ValueError("Duplicate link IDs detected")
+
+        # Rule: link endpoints must reference existing nodes
+        for lk in self.links:
+            if lk.source not in node_ids:
+                raise ValueError(f"Link '{lk.id}' source '{lk.source}' not in nodes")
+            if lk.target not in node_ids:
+                raise ValueError(f"Link '{lk.id}' target '{lk.target}' not in nodes")
+
+        return self

web/app/services/opm_extract.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+
+def extract_opm_diagram(text: str) -> dict:
+    """
+    Phase 1 stub implementation.
+
+    Ignores input text and returns a deterministic hardcoded diagram.
+    This function will be replaced by an LLM-backed implementation in Phase 2.
+    """
+    return {
+        "version": "1.0",
+        "nodes": [
+            {"id": "example-object", "kind": "object", "label": "example"}
+        ],
+        "links": [],
+    }

web/app/services/opm_validate.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+import logging
+
+from pydantic import ValidationError  # noqa: F401 — re-exported for callers
+
+from ..schemas.opm import OpmDiagram
+
+logger = logging.getLogger(__name__)
+
+
+def validate_diagram(data: dict) -> OpmDiagram:
+    """
+    Validate a candidate diagram dict.
+
+    Returns:
+        OpmDiagram on success
+
+    Raises:
+        pydantic.ValidationError on schema/field failures
+        ValueError on graph-integrity failures
+    """
+    diagram = OpmDiagram.model_validate(data)
+    _warn_semantic(diagram)
+    return diagram
+
+
+def _warn_semantic(diagram: OpmDiagram) -> None:
+    # Self-loops
+    for lk in diagram.links:
+        if lk.source == lk.target:
+            logger.warning("Self-loop on node '%s' via link '%s'", lk.source, lk.id)
+
+    # Duplicate (source, target, relation) triples
+    seen: set[tuple] = set()
+    for lk in diagram.links:
+        key = (lk.source, lk.target, lk.relation)
+        if key in seen:
+            logger.warning(
+                "Duplicate relation %s on (%s → %s)",
+                lk.relation,
+                lk.source,
+                lk.target,
+            )
+        seen.add(key)