Feat/posthog integration

.github/workflows/azure-static-web-apps-jolly-ocean-0fcd5df0f.yml

@@ -33,6 +33,8 @@ jobs:
         run: |
           echo "VITE_API_BASE=${{ secrets.VITE_API_BASE }}" >> .env
           echo "VITE_CLERK_PUBLISHABLE_KEY=${{ secrets.VITE_CLERK_PUBLISHABLE_KEY }}" >> .env
+          echo "VITE_POSTHOG_KEY=${{ secrets.VITE_POSTHOG_KEY }}" >> .env
+          echo "VITE_POSTHOG_HOST=${{ secrets.VITE_POSTHOG_HOST }}" >> .env
 
       - name: Install
         working-directory: packages/frontend

packages/backend/.env.example

@@ -9,3 +9,10 @@ INNGEST_SIGNING_KEY=
 CLERK_JWKS_URL=https://<your-clerk-frontend-api>.clerk.accounts.dev/.well-known/jwks.json
 # Comma-separated list of allowed frontend origins (matches the `azp` JWT claim)
 CLERK_AUTHORIZED_PARTIES=["http://localhost:5173"]
+
+# PostHog analytics (optional — leave POSTHOG_ENABLED=false to disable entirely)
+POSTHOG_API_KEY=
+POSTHOG_HOST=https://us.i.posthog.com
+POSTHOG_ENABLED=false
+
+POSTHOG_REDACT_PROMPTS=true

packages/backend/app/ai/core/__init__.py

@@ -1,4 +1,4 @@
-from app.ai.core.client import DEFAULT_MODEL, GUARDRAIL_MODEL, client
+from app.ai.core.client import DEFAULT_MODEL, GUARDRAIL_MODEL, client, posthog_client
 from app.ai.core.output_cleaner import (
     clean_section_output,
     promote_h1_headings,
@@ -18,6 +18,7 @@
 
 __all__ = [
     "client",
+    "posthog_client",
     "DEFAULT_MODEL",
     "GUARDRAIL_MODEL",
     "truncate_section",

packages/backend/app/ai/core/client.py

@@ -1,10 +1,16 @@
-from openai import AsyncOpenAI
+from posthog import Posthog
+from posthog.ai.openai import AsyncOpenAI
 
 from app.config import settings
 
+posthog_client: Posthog | None = None
+if settings.POSTHOG_ENABLED and settings.POSTHOG_API_KEY:
+    posthog_client = Posthog(settings.POSTHOG_API_KEY, host=settings.POSTHOG_HOST)
+
 client = AsyncOpenAI(
     base_url=settings.AZURE_OPENAI_ENDPOINT,
-    api_key=settings.AZURE_OPENAI_API_KEY
+    api_key=settings.AZURE_OPENAI_API_KEY,
+    posthog_client=posthog_client,
 )
 DEFAULT_MODEL = settings.AZURE_OPENAI_DEPLOYMENT
 GUARDRAIL_MODEL = settings.AZURE_OPENAI_GUARDRAIL_DEPLOYMENT

packages/backend/app/config.py

@@ -30,6 +30,12 @@ class Settings(BaseSettings):
     GUIDED_DOCUMENT_COST: int = 3
     EDITOR_DOCUMENT_COST: int = 1
 
+    # PostHog analytics — all disabled by default; set POSTHOG_ENABLED=true to activate
+    POSTHOG_API_KEY: str = ""
+    POSTHOG_HOST: str = "https://us.i.posthog.com"
+    POSTHOG_ENABLED: bool = False
+    POSTHOG_REDACT_PROMPTS: bool = True
+
     model_config = {"env_file": ENV_FILE, "extra": "ignore"}
 
 

packages/backend/app/guardrails/input.py

@@ -103,7 +103,13 @@ def validate_refinement_message(message: str) -> ValidationError | None:
 }
 
 
-async def validate_message_intent(message: str, section_type: str, section_summary: str | None = None) -> ValidationError | None:
+async def validate_message_intent(
+    message: str,
+    section_type: str,
+    section_summary: str | None = None,
+    posthog_distinct_id: str | None = None,
+    posthog_properties: dict | None = None,
+) -> ValidationError | None:
     """LLM-based intent classifier: rejects messages unrelated to document editing.
 
     Failures are swallowed — a guardrail error must never block the user.
@@ -115,18 +121,23 @@ async def validate_message_intent(message: str, section_type: str, section_summa
     system_prompt = load_yaml_prompt("guardrails", "intent_classification", "system")
 
     try:
-        response = await client.chat.completions.create(
-            model=GUARDRAIL_MODEL,
-            messages=[
+        create_kwargs: dict = {
+            "model": GUARDRAIL_MODEL,
+            "messages": [
                 {"role": "system", "content": system_prompt},
                 {
                     "role": "user",
                     "content": f"Section: {section_type}\nSummary: {section_summary}\nMessage: {message}",
                 },
             ],
-            response_format=_INTENT_SCHEMA,
-            temperature=0.0,
-        )
+            "response_format": _INTENT_SCHEMA,
+            "temperature": 1.0,
+        }
+        if posthog_distinct_id is not None:
+            create_kwargs["posthog_distinct_id"] = posthog_distinct_id
+        if posthog_properties is not None:
+            create_kwargs["posthog_properties"] = posthog_properties
+        response = await client.chat.completions.create(**create_kwargs)
         result = json.loads(response.choices[0].message.content)
         if not result.get("is_valid", True):
             logger.warning(

packages/backend/app/guardrails/output.py

@@ -17,6 +17,8 @@ async def call_with_retry(
     temperature: float = 0.3,
     required_fields: list[str] | None = None,
     section_type: str | None = None,
+    posthog_distinct_id: str | None = None,
+    posthog_properties: dict | None = None,
 ) -> Any:
     """
     Call the AI model and validate the structured response.
@@ -37,6 +39,10 @@ async def call_with_retry(
         kwargs["tools"] = tools
     if tool_choice is not None:
         kwargs["tool_choice"] = tool_choice
+    if posthog_distinct_id is not None:
+        kwargs["posthog_distinct_id"] = posthog_distinct_id
+    if posthog_properties is not None:
+        kwargs["posthog_properties"] = posthog_properties
 
     response = await client.chat.completions.create(**kwargs)
     error = _validate_response(response, required_fields)

packages/backend/app/phases/alignment/ai.py

@@ -60,6 +60,8 @@ async def generate_alignment(
     rejected_sections: list[dict] | None = None,
     db: AsyncSession | None = None,
     document_type_id: uuid.UUID | None = None,
+    posthog_distinct_id: str | None = None,
+    doc_id: str | None = None,
 ) -> dict:
     """Generate 1-2 sentence summaries for each section using per-section discovery contexts."""
     logger.info(
@@ -85,6 +87,11 @@ async def generate_alignment(
         response_format=ALIGNMENT_SCHEMA,
         temperature=0.3,
         required_fields=["summaries"],
+        posthog_distinct_id=posthog_distinct_id,
+        posthog_properties=(
+            {"$ai_trace_id": doc_id, "$ai_parent_id": f"alignment-{doc_id}"}
+            if doc_id else None
+        ),
     )
 
     log_usage("alignment", response.usage)

packages/backend/app/phases/alignment/contract.py

@@ -54,6 +54,8 @@ async def extract_document_contract(
     user_preferences: str | None = None,
     db: AsyncSession | None = None,
     document_type_id: uuid.UUID | None = None,
+    posthog_distinct_id: str | None = None,
+    doc_id: str | None = None,
 ) -> dict:
     """Extract a structured document contract from the approved alignment summaries."""
     logger.info("[AI:contract] extracting document contract")
@@ -83,6 +85,11 @@ async def extract_document_contract(
         response_format=CONTRACT_SCHEMA,
         temperature=0.1,
         required_fields=["entities", "decisions", "terminology", "constraints"],
+        posthog_distinct_id=posthog_distinct_id,
+        posthog_properties=(
+            {"$ai_trace_id": doc_id, "$ai_parent_id": f"alignment-{doc_id}"}
+            if doc_id else None
+        ),
     )
 
     log_usage("contract", response.usage)

packages/backend/app/phases/alignment/function.py

@@ -9,6 +9,7 @@
 from app.phases._shared.concurrency import DOC_CONCURRENCY
 from app.phases._shared.failure import workflow_on_failure
 from app.services import db as db_service
+from app.services.observability import capture_event, capture_span
 
 
 @inngest_client.create_function(
@@ -21,6 +22,7 @@ async def function(ctx: inngest.Context):
     step = ctx.step
     doc_id = ctx.event.data["document_id"]
     document_type_id = ctx.event.data.get("document_type_id")
+    user_id = ctx.event.data.get("user_id", "")
 
     logger.info("[orchestrator] ALIGNMENT start | doc_id={}", doc_id)
 
@@ -66,6 +68,8 @@ async def _generate_summaries(sc=section_contexts):
                     None,
                     db=db,
                     document_type_id=uuid.UUID(document_type_id) if document_type_id else None,
+                    posthog_distinct_id=user_id,
+                    doc_id=doc_id,
                 )
 
         summaries = await step.run(f"generate-summaries-{iteration}", _generate_summaries)
@@ -105,6 +109,8 @@ async def _extract_contract(s=summaries, sc=section_contexts):
                         user_preferences=doc.user_preferences,
                         db=db,
                         document_type_id=uuid.UUID(document_type_id) if document_type_id else None,
+                        posthog_distinct_id=user_id,
+                        doc_id=doc_id,
                     )
                 import json as _json
                 raw = _json.dumps(contract)
@@ -113,17 +119,35 @@ async def _extract_contract(s=summaries, sc=section_contexts):
 
             await step.run("extract-contract", _extract_contract)
         else:
+            rejected = approval_event.data.get("rejected", [])
             logger.info(
                 "[orchestrator] alignment rejected sections={} | doc_id={}",
-                approval_event.data.get("rejected", []),
+                rejected,
                 doc_id,
             )
 
+            async def _capture_rejected(r=rejected, it=iteration):
+                for section_type in r:
+                    capture_event(user_id, doc_id, "alignment_section_reopened", {"section_type": section_type, "iteration": it})
+
+            await step.run(f"capture-alignment-rejected-{iteration}", _capture_rejected)
+
+    async def _capture_phase_span():
+        capture_span(
+            user_id, doc_id,
+            span_id=f"alignment-{doc_id}",
+            span_name="alignment",
+            input_state="Section contexts reviewed",
+            output_state="Alignment approved",
+        )
+
+    await step.run("capture-span-alignment", _capture_phase_span)
+
     logger.info("[orchestrator] emitting alignment.completed | doc_id={}", doc_id)
     await step.send_event(
         "emit-alignment-completed",
         inngest.Event(
             name="docforge/alignment.completed",
-            data={"document_id": doc_id, "document_type_id": document_type_id},
+            data={"document_id": doc_id, "document_type_id": document_type_id, "user_id": user_id},
         ),
     )

packages/backend/app/phases/audit/ai.py

@@ -88,7 +88,7 @@ def build_audit_context(
     return "\n\n".join(parts)
 
 
-async def run_audit(doc_id: str, document_type_id: str | None = None) -> dict:
+async def run_audit(doc_id: str, document_type_id: str | None = None, posthog_distinct_id: str | None = None) -> dict:
     """Run audit on all 4 finalized sections."""
     logger.info("[AI:audit] run_audit | doc_id={}", doc_id)
     from app.services import db as db_service
@@ -129,6 +129,11 @@ async def run_audit(doc_id: str, document_type_id: str | None = None) -> dict:
         response_format=AUDIT_SCHEMA,
         temperature=0.2,
         required_fields=["has_problems", "problems"],
+        posthog_distinct_id=posthog_distinct_id,
+        posthog_properties=(
+            {"$ai_trace_id": doc_id, "$ai_parent_id": f"audit-{doc_id}"}
+            if posthog_distinct_id else None
+        ),
     )
 
     log_usage("audit", response.usage)

packages/backend/app/phases/audit/function.py

@@ -8,6 +8,7 @@
 from app.phases._shared.concurrency import DOC_CONCURRENCY
 from app.phases._shared.failure import workflow_on_failure
 from app.services import db as db_service
+from app.services.observability import capture_span
 
 
 @inngest_client.create_function(
@@ -20,6 +21,7 @@ async def function(ctx: inngest.Context):
     step = ctx.step
     doc_id = ctx.event.data["document_id"]
     document_type_id = ctx.event.data.get("document_type_id")
+    user_id = ctx.event.data.get("user_id", "")
 
     logger.info("[orchestrator] AUDIT start | doc_id={}", doc_id)
 
@@ -31,7 +33,7 @@ async def _set_audit():
 
     async def _run_audit():
         from app.phases.audit.ai import run_audit
-        return await run_audit(doc_id, document_type_id=document_type_id)
+        return await run_audit(doc_id, document_type_id=document_type_id, posthog_distinct_id=user_id)
 
     result = await step.run("run-audit", _run_audit)
 
@@ -55,10 +57,23 @@ async def _save_findings(r=result):
 
     await step.run("save-findings", _save_findings)
 
+    async def _capture_phase_span(r=result):
+        problems = r.get("problems", [])
+        output = f"Audit complete. {len(problems)} issue(s) found." if problems else "Audit complete. No issues found."
+        capture_span(
+            user_id, doc_id,
+            span_id=f"audit-{doc_id}",
+            span_name="audit",
+            input_state="Document sections reviewed",
+            output_state=output,
+        )
+
+    await step.run("capture-span-audit", _capture_phase_span)
+
     await step.send_event(
         "emit-audit-completed",
         inngest.Event(
             name="docforge/document.audit_completed",
-            data={"document_id": doc_id, "document_type_id": document_type_id},
+            data={"document_id": doc_id, "document_type_id": document_type_id, "user_id": user_id},
         ),
     )

packages/backend/app/phases/completion/function.py

@@ -6,6 +6,7 @@
 from app.phases._shared.concurrency import DOC_CONCURRENCY
 from app.phases._shared.failure import workflow_on_failure
 from app.services import db as db_service
+from app.services.observability import capture_trace
 
 
 @inngest_client.create_function(
@@ -17,6 +18,7 @@
 async def function(ctx: inngest.Context):
     step = ctx.step
     doc_id = ctx.event.data["document_id"]
+    user_id = ctx.event.data.get("user_id", "")
 
     logger.info("[orchestrator] COMPLETED | doc_id={}", doc_id)
 
@@ -25,3 +27,8 @@ async def _set_completed():
             await db_service.set_phase(db, doc_id, "completed")
 
     await step.run("set-phase-completed", _set_completed)
+
+    async def _close_trace():
+        capture_trace(user_id, doc_id, output_state="Document complete")
+
+    await step.run("capture-trace-complete", _close_trace)

packages/backend/app/phases/discovery/ai.py

@@ -59,6 +59,8 @@ async def analyze_discovery(
     section_role: str,
     db: AsyncSession,
     document_type_id: uuid.UUID | None,
+    posthog_distinct_id: str | None = None,
+    doc_id: str | None = None,
 ) -> dict:
     """Analyze if context is sufficient for a specific section, or generate follow-up questions."""
     logger.info(
@@ -80,6 +82,11 @@ async def analyze_discovery(
         response_format=DISCOVERY_SCHEMA,
         temperature=0.3,
         required_fields=["is_sufficient", "follow_up_questions"],
+        posthog_distinct_id=posthog_distinct_id,
+        posthog_properties=(
+            {"$ai_trace_id": doc_id, "$ai_parent_id": f"discovery-{section_key}-{doc_id}"}
+            if doc_id else None
+        ),
     )
 
     log_usage("discovery", response.usage)