diff --git a/.env.example b/.env.example index 1314317..1f43460 100644 --- a/.env.example +++ b/.env.example @@ -6,6 +6,12 @@ HOST_PORT=8011 # Optional: override the Anthropic model used for extraction ANTHROPIC_MODEL=claude-sonnet-4-6 +# Durable CSV copies of discovery questions + answers. The per-conversation +# files live in QUESTIONS_DIR; QUESTIONS_MASTER_CSV is an append-only log of +# every save/complete event across conversations. Both default under ./data. +QUESTIONS_DIR=./data/questions +QUESTIONS_MASTER_CSV=./data/questions_master.csv + # Google OAuth (web client type). Register the redirect URI below as an # authorized redirect URI in the Google Cloud Console for this client. GOOGLE_CLIENT_ID=xxxx.apps.googleusercontent.com diff --git a/.gitignore b/.gitignore index 7a71767..3a634b8 100644 --- a/.gitignore +++ b/.gitignore @@ -7,5 +7,7 @@ data/*.db data/*.db-journal data/*.log data/*.err.log +data/questions/ +data/questions_master.csv .pytest_cache/ *.egg-info/ diff --git a/README.md b/README.md index a8a7d27..da15782 100644 --- a/README.md +++ b/README.md @@ -196,6 +196,7 @@ Important files: | `app/models.py` | SQLAlchemy ORM models for users, refresh tokens, activity log, conversations, and profiles | | `app/database.py` | Async database engine, session factory, SQLite directory setup | | `app/services/extractor.py` | Anthropic client wrapper, prompt, JSON parsing, retry logic | +| `app/services/answer_store.py` | Durable CSV persistence of discovery questions/answers (per-conversation + master log); canonical prompt list | | `app/migration_bootstrap.py` | Stamps pre-Alembic SQLite DBs as revision `001` so `alembic upgrade head` succeeds on older local databases | | `app/static/discovery.html` | Browser-based seven-prompt flow | | `app/static/profile.html` | Browser-based profile display, edit, and confirm actions; links to reflection | @@ -212,6 +213,7 @@ Important files: | `app/services/tagging.py` | `FoundationTagger`: Anthropic smart-tagging suggestion (Phase 5) | | `app/routers/coaching.py` | Phase 3 coaching routes: preferences, check-ins, weekly batch `/run` | | `app/routers/integration.py` | Phase 4/5 integration: foundations, task-mappings, work-patterns, suggest-foundation | +| `scripts/reprocess_csv.py` | Offline CLI to re-run AI extraction over a saved answers CSV (print or `--write-db`) | | `alembic/versions/001_initial.py` | Initial database schema migration | | `alembic/versions/002_add_auth.py` | Adds `users`, `refresh_tokens`, and `activity_log` tables | | `alembic/versions/003_add_goals.py` | Adds the goal columns to `discovery_conversation` and `discovery_profile` | @@ -229,6 +231,8 @@ Important files: | `tests/test_phase5.py` | Tests for goal-history and smart-tagging endpoints | | `tests/test_auth.py` | Tests for the dual-auth dependency (JWT + cookie + API key), token refresh/logout, admin enforcement, and domain allow-list | | `tests/test_profile_edit.py` | Tests for `PATCH /discovery/profile/me` (edit, partial update, lock/`409`) | +| `tests/test_answer_store.py` | Unit tests for the CSV answer store (round-trip, atomic overwrite, master header, fallbacks) | +| `tests/test_discovery_csv.py` | Tests for the CSV/reprocess endpoints (respond writes CSV, answers, download, reprocess, lock/`409`) | | `tests/test_reflection.py` | Tests for the reflection endpoints (turns, applied revisions, lock/`409`, history) | | `tests/test_migration_bootstrap.py` | Unit tests for the pre-Alembic SQLite stamping helper | | `tests/test_static_discovery.py` | Guard tests for the static pages' cookie-session and edit contract | @@ -393,6 +397,10 @@ It stores the responses on the existing conversation and returns: If the conversation id does not exist, it returns `404`. +On every save the answers are also written to a durable CSV copy (see +[Answer CSV persistence](#answer-csv-persistence)) so they survive an +extraction error and can be reviewed or re-processed. + ### 4. Completing Analysis `POST /discovery/{conversation_id}/complete` loads the conversation, builds a @@ -409,6 +417,55 @@ The route rejects completion with: On success, it stores a new `DiscoveryProfile`, marks the conversation `completed_at`, and returns the profile. +### Answer CSV persistence + +Every discovery conversation's prompts and answers are mirrored to CSV on +disk so the answers are durable beyond the database, recoverable after an +extraction error, and re-feedable to the AI. The SQLite database stays the +system of record — CSV write failures are logged and never break a request. + +Two artifacts are written, on both save (`/respond`) and completion +(`/complete`): + +- **Per-conversation file** — `data/questions/{conversation_id}.csv`, + rewritten in full on each save (latest answers, atomic write). +- **Master log** — `data/questions_master.csv`, an append-only record of + every save/complete event across all conversations, for batch + re-processing. + +Both use a long format (one row per prompt) with columns: +`conversation_id, user_id, user_email, status, saved_at, prompt_key, +prompt_title, answer`. Paths are configurable via `QUESTIONS_DIR` and +`QUESTIONS_MASTER_CSV`. + +Related endpoints: + +- `GET /discovery/{conversation_id}/answers` — saved prompts + answers as + JSON (used by the browser flow to resume/review original answers). +- `GET /discovery/{conversation_id}/answers.csv` — download the answers as a + CSV file (built from the DB, so it works even if the on-disk copy is + missing). +- `POST /discovery/{conversation_id}/reprocess` — re-run extraction over the + saved answers, producing a fresh profile. Returns `409` if the latest + profile is locked, `400` if there is nothing to analyze, `502` on an + extractor failure. The discovery page's error screen offers a **Re-run + analysis** button wired to this endpoint. + +Offline/bulk re-processing is available via a CLI that needs no running +server: + +```bash +# Print the regenerated profile JSON for inspection (default, no DB write): +python scripts/reprocess_csv.py data/questions/.csv + +# Re-process a single conversation out of the master log and persist it: +python scripts/reprocess_csv.py data/questions_master.csv \ + --conversation-id --write-db +``` + +It reads `ANTHROPIC_API_KEY` (and optional `ANTHROPIC_MODEL`) from the +environment. + ### 5. Loading The Profile `GET /discovery/profile/me` fetches the newest profile for the authenticated diff --git a/app/routers/discovery.py b/app/routers/discovery.py index 292ef4c..10eeaad 100644 --- a/app/routers/discovery.py +++ b/app/routers/discovery.py @@ -11,6 +11,7 @@ import uuid from datetime import datetime, timezone from fastapi import APIRouter, Depends, HTTPException +from fastapi.responses import Response from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession @@ -24,6 +25,8 @@ from app.models import ( ReflectionMessage, User, ) +from app.services import answer_store +from app.services.answer_store import DISCOVERY_PROMPTS from app.services.extractor import DiscoveryExtractionError, DiscoveryExtractor from app.services.reflector import ( EDITABLE_FIELDS, @@ -66,6 +69,60 @@ def _record_revision( ) +async def _generate_profile( + db: AsyncSession, + conversation: DiscoveryConversation, + source: str, +) -> tuple[DiscoveryProfile, str | None]: + """Run the extractor over a conversation's answers and stage a new profile + (plus a revision snapshot) on the session. The caller commits. + + Shared by ``/complete`` (source="extraction") and ``/reprocess`` + (source="reprocess"). Raises HTTPException(400) when there is nothing to + analyze and HTTPException(502) on an extractor failure. + """ + responses = { + p["key"]: getattr(conversation, p["column"], None) or "" + for p in DISCOVERY_PROMPTS + } + if not any(text.strip() for text in responses.values()): + raise HTTPException( + status_code=400, detail="No responses available to analyze" + ) + + api_key = os.getenv("ANTHROPIC_API_KEY") + model = os.getenv("ANTHROPIC_MODEL", "claude-sonnet-4-6") + try: + extractor = DiscoveryExtractor(api_key=api_key, model=model) + data = await extractor.extract(responses) + except DiscoveryExtractionError as exc: + raise HTTPException(status_code=502, detail=str(exc)) from exc + + profile = DiscoveryProfile( + id=str(uuid.uuid4()), + user_id=conversation.user_id, + conversation_id=conversation.id, + generated_at=_now(), + triad=data.get("triad"), + probable_type=_as_int(data.get("probable_type")), + wing=_as_int(data.get("wing")), + instinctual_variant=data.get("instinctual_variant"), + instinctual_stack=data.get("instinctual_stack"), + love_summary=data.get("love_summary"), + strength_summary=data.get("strength_summary"), + mission_summary=data.get("mission_summary"), + vocation_summary=data.get("vocation_summary"), + overlap_narrative=data.get("overlap_narrative"), + short_term_goals=data.get("short_term_goals"), + long_term_goals=data.get("long_term_goals"), + confidence_json=json.dumps(data.get("confidence", {})), + locked=False, + ) + db.add(profile) + _record_revision(db, profile, source=source) + return profile, data.get("extraction_notes") + + def _to_profile_response( profile: DiscoveryProfile, extraction_notes: str | None = None ) -> schemas.ProfileResponse: @@ -161,6 +218,10 @@ async def save_responses( conversation.prompt_goals_long = payload.prompt_goals_long await db.commit() + # Durable CSV copy, written before extraction runs so the answers survive + # an extraction error and can be reviewed or re-processed later. + answer_store.save_answers(conversation, user, status="responses_saved") + return schemas.RespondResponse( conversation_id=conversation_id, status="responses_saved" ) @@ -176,59 +237,99 @@ async def complete_conversation( ): conversation = await _owned_conversation(db, conversation_id, user) - responses = { - "alive": conversation.prompt_alive or "", - "friction": conversation.prompt_friction or "", - "pull": conversation.prompt_pull or "", - "recognition": conversation.prompt_recognition or "", - "future": conversation.prompt_future or "", - "goals_short": conversation.prompt_goals_short or "", - "goals_long": conversation.prompt_goals_long or "", - } - if not any(text.strip() for text in responses.values()): - raise HTTPException( - status_code=400, detail="No responses available to analyze" - ) - - api_key = os.getenv("ANTHROPIC_API_KEY") - model = os.getenv("ANTHROPIC_MODEL", "claude-sonnet-4-6") - - try: - extractor = DiscoveryExtractor(api_key=api_key, model=model) - data = await extractor.extract(responses) - except DiscoveryExtractionError as exc: - raise HTTPException(status_code=502, detail=str(exc)) from exc - - profile = DiscoveryProfile( - id=str(uuid.uuid4()), - user_id=conversation.user_id, - conversation_id=conversation.id, - generated_at=_now(), - triad=data.get("triad"), - probable_type=_as_int(data.get("probable_type")), - wing=_as_int(data.get("wing")), - instinctual_variant=data.get("instinctual_variant"), - instinctual_stack=data.get("instinctual_stack"), - love_summary=data.get("love_summary"), - strength_summary=data.get("strength_summary"), - mission_summary=data.get("mission_summary"), - vocation_summary=data.get("vocation_summary"), - overlap_narrative=data.get("overlap_narrative"), - short_term_goals=data.get("short_term_goals"), - long_term_goals=data.get("long_term_goals"), - confidence_json=json.dumps(data.get("confidence", {})), - locked=False, + profile, extraction_notes = await _generate_profile( + db, conversation, source="extraction" ) conversation.completed_at = _now() - db.add(profile) - _record_revision(db, profile, source="extraction") await db.commit() - return _to_profile_response( - profile, extraction_notes=data.get("extraction_notes") + # Refresh the durable CSV copy now that the conversation is complete. + answer_store.save_answers(conversation, user, status="completed") + + return _to_profile_response(profile, extraction_notes=extraction_notes) + + +@router.get( + "/{conversation_id}/answers", response_model=schemas.AnswersResponse +) +async def get_answers( + conversation_id: str, + db: AsyncSession = Depends(get_db), + user: User = Depends(get_current_user), +): + """The saved prompts and answers for a conversation, so the person can + review or resume from their original responses (e.g. after an error).""" + conversation = await _owned_conversation(db, conversation_id, user) + answers = [ + schemas.AnswerItem( + prompt_key=p["key"], + prompt_title=p["title"], + answer=getattr(conversation, p["column"], None) or "", + ) + for p in DISCOVERY_PROMPTS + ] + return schemas.AnswersResponse( + conversation_id=conversation.id, + started_at=conversation.started_at, + completed_at=conversation.completed_at, + answers=answers, ) +@router.get("/{conversation_id}/answers.csv") +async def download_answers_csv( + conversation_id: str, + db: AsyncSession = Depends(get_db), + user: User = Depends(get_current_user), +): + """Download a conversation's answers as a CSV file (built from the DB so it + works even if the on-disk copy was never written).""" + conversation = await _owned_conversation(db, conversation_id, user) + status = "completed" if conversation.completed_at else "responses_saved" + csv_text = answer_store.conversation_csv_text(conversation, user, status) + filename = f"discovery-{conversation_id}.csv" + return Response( + content=csv_text, + media_type="text/csv", + headers={ + "Content-Disposition": f'attachment; filename="{filename}"' + }, + ) + + +@router.post( + "/{conversation_id}/reprocess", response_model=schemas.ProfileResponse +) +async def reprocess_conversation( + conversation_id: str, + db: AsyncSession = Depends(get_db), + user: User = Depends(get_current_user), +): + """Re-run AI extraction over a conversation's saved answers, producing a + fresh profile. Used to recover from an extraction error or to regenerate a + profile after the answers were re-fed. The latest profile must be unlocked. + """ + conversation = await _owned_conversation(db, conversation_id, user) + + existing = await _latest_profile(db, user.id) + if existing is not None and existing.locked: + raise HTTPException( + status_code=409, + detail="Profile is affirmed and locked; it cannot be reprocessed.", + ) + + profile, extraction_notes = await _generate_profile( + db, conversation, source="reprocess" + ) + if conversation.completed_at is None: + conversation.completed_at = _now() + await db.commit() + + answer_store.save_answers(conversation, user, status="completed") + + return _to_profile_response(profile, extraction_notes=extraction_notes) + + @router.get("/profile/me", response_model=schemas.ProfileResponse) async def get_my_profile( db: AsyncSession = Depends(get_db), diff --git a/app/schemas.py b/app/schemas.py index 8e9beb2..124ee51 100644 --- a/app/schemas.py +++ b/app/schemas.py @@ -24,6 +24,21 @@ class RespondResponse(BaseModel): status: str +class AnswerItem(BaseModel): + prompt_key: str + prompt_title: str + answer: str = "" + + +class AnswersResponse(BaseModel): + """The saved prompts and answers for a conversation, for review/resume.""" + + conversation_id: str + started_at: datetime + completed_at: Optional[datetime] = None + answers: list[AnswerItem] = [] + + class Confidence(BaseModel): triad: Optional[str] = None type: Optional[str] = None diff --git a/app/services/answer_store.py b/app/services/answer_store.py new file mode 100644 index 0000000..bfb15f1 --- /dev/null +++ b/app/services/answer_store.py @@ -0,0 +1,223 @@ +"""Durable CSV record of discovery questions and answers. + +The SQLite database remains the system of record. This module keeps a +secondary, portable copy of every conversation's prompts and answers on disk +so that: + + * answers survive even if the AI extraction step errors (they are written + on save, before extraction runs); + * a saved conversation can be re-fed to the extractor (``reprocess``); + * a person can review or resume from their original answers. + +Two artifacts are written for every save: + + * a *per-conversation* file (``data/questions/{conversation_id}.csv``) that + is rewritten in full on each save — always the latest answers for that + conversation, easy to hand to an AI or download; and + * a *master* append-only log (``data/questions_master.csv``) that records + every save/complete event across all conversations, for batch + re-processing. + +CSV writes must never break an API request: the caller's DB commit has +already succeeded, so any I/O failure here is logged and swallowed. +""" +from __future__ import annotations + +import csv +import logging +import os +from datetime import datetime, timezone +from typing import TYPE_CHECKING + +if TYPE_CHECKING: # pragma: no cover - typing only + from app.models import DiscoveryConversation, User + +logger = logging.getLogger(__name__) + +# Canonical list of discovery prompts: the short key used by the extractor and +# CSV, the DB column on DiscoveryConversation, and the human-facing title. +# This is the single source of truth shared by the store, the router, and the +# reprocess paths. +DISCOVERY_PROMPTS = [ + {"key": "alive", "column": "prompt_alive", "title": "The Alive Moment"}, + {"key": "friction", "column": "prompt_friction", "title": "The Friction Moment"}, + {"key": "pull", "column": "prompt_pull", "title": "The Natural Pull"}, + {"key": "recognition", "column": "prompt_recognition", "title": "The Recognition Moment"}, + {"key": "future", "column": "prompt_future", "title": "The Future Pull"}, + {"key": "goals_short", "column": "prompt_goals_short", "title": "Near-Term Goals (6–12 months)"}, + {"key": "goals_long", "column": "prompt_goals_long", "title": "Long-Term Goals (3–5 years)"}, +] + +CSV_FIELDS = [ + "conversation_id", + "user_id", + "user_email", + "status", + "saved_at", + "prompt_key", + "prompt_title", + "answer", +] + + +def _questions_dir() -> str: + return os.getenv("QUESTIONS_DIR", "./data/questions") + + +def _master_path() -> str: + return os.getenv("QUESTIONS_MASTER_CSV", "./data/questions_master.csv") + + +def conversation_csv_path(conversation_id: str) -> str: + """Filesystem path of the per-conversation CSV for ``conversation_id``.""" + return os.path.join(_questions_dir(), f"{conversation_id}.csv") + + +def _rows_for( + conversation: "DiscoveryConversation", + user_email: str, + status: str, + saved_at: str, +) -> list[dict]: + """One row per prompt, in canonical prompt order.""" + rows = [] + for prompt in DISCOVERY_PROMPTS: + answer = getattr(conversation, prompt["column"], None) or "" + rows.append( + { + "conversation_id": conversation.id, + "user_id": conversation.user_id, + "user_email": user_email, + "status": status, + "saved_at": saved_at, + "prompt_key": prompt["key"], + "prompt_title": prompt["title"], + "answer": answer, + } + ) + return rows + + +def write_conversation_csv( + conversation: "DiscoveryConversation", + user: "User | None", + status: str, +) -> bool: + """Rewrite the per-conversation CSV with the conversation's current answers. + + The write is atomic (temp file + ``os.replace``) so a crash mid-write can + never leave a half-written file. Returns True on success, False if the + write failed (failures are logged, never raised). + """ + saved_at = datetime.now(timezone.utc).isoformat() + email = getattr(user, "email", "") or "" + rows = _rows_for(conversation, email, status, saved_at) + path = conversation_csv_path(conversation.id) + tmp_path = f"{path}.tmp" + try: + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(tmp_path, "w", newline="", encoding="utf-8") as fh: + writer = csv.DictWriter(fh, fieldnames=CSV_FIELDS) + writer.writeheader() + writer.writerows(rows) + os.replace(tmp_path, path) + return True + except OSError as exc: + logger.warning( + "Failed to write conversation CSV for %s: %s", conversation.id, exc + ) + try: + if os.path.exists(tmp_path): + os.remove(tmp_path) + except OSError: + pass + return False + + +def append_master( + conversation: "DiscoveryConversation", + user: "User | None", + status: str, +) -> bool: + """Append this save/complete event's rows to the master log. + + Writes the header row once, when the file is first created. Returns True on + success, False on a logged (never raised) failure. + """ + saved_at = datetime.now(timezone.utc).isoformat() + email = getattr(user, "email", "") or "" + rows = _rows_for(conversation, email, status, saved_at) + path = _master_path() + try: + directory = os.path.dirname(path) + if directory: + os.makedirs(directory, exist_ok=True) + is_new = not os.path.exists(path) or os.path.getsize(path) == 0 + with open(path, "a", newline="", encoding="utf-8") as fh: + writer = csv.DictWriter(fh, fieldnames=CSV_FIELDS) + if is_new: + writer.writeheader() + writer.writerows(rows) + return True + except OSError as exc: + logger.warning( + "Failed to append master CSV for %s: %s", conversation.id, exc + ) + return False + + +def save_answers( + conversation: "DiscoveryConversation", + user: "User | None", + status: str, +) -> None: + """Persist both CSV artifacts for a conversation. Never raises.""" + write_conversation_csv(conversation, user, status) + append_master(conversation, user, status) + + +def conversation_csv_text( + conversation: "DiscoveryConversation", + user: "User | None", + status: str = "responses_saved", +) -> str: + """Render a conversation's answers as CSV text, built from the DB row. + + Used by the download endpoint so it works even if the on-disk file was + never written (DB stays the system of record). + """ + import io + + saved_at = datetime.now(timezone.utc).isoformat() + email = getattr(user, "email", "") or "" + rows = _rows_for(conversation, email, status, saved_at) + buf = io.StringIO() + writer = csv.DictWriter(buf, fieldnames=CSV_FIELDS) + writer.writeheader() + writer.writerows(rows) + return buf.getvalue() + + +def read_conversation_answers(conversation_id: str) -> dict[str, str]: + """Read saved answers from the per-conversation CSV. + + Returns a ``{prompt_key: answer}`` dict, or ``{}`` if the file is missing + or unreadable. The per-conversation file is rewritten in full on each save, + so it always holds the latest answers. + """ + path = conversation_csv_path(conversation_id) + answers: dict[str, str] = {} + try: + with open(path, newline="", encoding="utf-8") as fh: + for row in csv.DictReader(fh): + key = row.get("prompt_key") + if key: + answers[key] = row.get("answer", "") or "" + except FileNotFoundError: + return {} + except OSError as exc: + logger.warning( + "Failed to read conversation CSV for %s: %s", conversation_id, exc + ) + return {} + return answers diff --git a/app/static/discovery.html b/app/static/discovery.html index cd3b75e..ca26f77 100644 --- a/app/static/discovery.html +++ b/app/static/discovery.html @@ -133,11 +133,62 @@ } }); + const STORAGE_KEY = "discovery_conversation_id"; + async function startConversation() { const res = await authedFetch("/discovery/start", { method: "POST" }); if (!res.ok) throw new Error("Could not start conversation"); const data = await res.json(); conversationId = data.conversation_id; + // Remember the in-progress conversation so a reload after an error can + // resume from the original answers rather than losing them. + try { + localStorage.setItem(STORAGE_KEY, conversationId); + } catch (e) { + /* storage unavailable — non-fatal */ + } + } + + async function resumeOrStart() { + let savedId = null; + try { + savedId = localStorage.getItem(STORAGE_KEY); + } catch (e) { + /* storage unavailable */ + } + if (savedId) { + try { + const res = await authedFetch(`/discovery/${savedId}/answers`); + if (res.ok) { + const data = await res.json(); + const hasText = (data.answers || []).some((a) => a.answer.trim()); + if (!data.completed_at && hasText) { + // Resume: restore the saved answers into the flow. + conversationId = savedId; + const byKey = {}; + data.answers.forEach((a) => { + byKey[a.prompt_key] = a.answer; + }); + PROMPTS.forEach((p, i) => { + // CSV/key uses the short key; PROMPTS uses prompt_* column names. + const shortKey = p.key.replace(/^prompt_/, ""); + answers[i] = byKey[shortKey] || ""; + }); + render(); + return; + } + } + } catch (e) { + /* fall through to a fresh start */ + } + } + // No resumable conversation — start a fresh one. + try { + localStorage.removeItem(STORAGE_KEY); + } catch (e) { + /* ignore */ + } + await startConversation(); } async function submit() { @@ -173,24 +224,70 @@ throw new Error(detail.detail || "Extraction failed"); } + // Success: the conversation is complete, drop the resume marker. + try { + localStorage.removeItem(STORAGE_KEY); + } catch (e) { + /* ignore */ + } window.location.href = "/static/profile.html"; } catch (err) { el.loading.classList.remove("active"); el.flow.style.display = "block"; el.flow.innerHTML = `
Something went wrong.
` + - `${err.message}

Your answers are still here — ` + - `please try submitting again.
` + + `${err.message}

Your answers are saved — you can ` + + `re-run the analysis or reload to keep editing.` + + `
` + + `` + + `` + + `
`; + const rerun = document.getElementById("rerunBtn"); + if (rerun) rerun.addEventListener("click", reprocess); + } + } + + async function reprocess() { + if (!conversationId) { + location.reload(); + return; + } + el.flow.style.display = "none"; + el.loading.classList.add("active"); + try { + const res = await authedFetch( + `/discovery/${conversationId}/reprocess`, + { method: "POST" } + ); + if (!res.ok) { + const detail = await res + .json() + .catch(() => ({ detail: "Re-run failed" })); + throw new Error(detail.detail || "Re-run failed"); + } + try { + localStorage.removeItem(STORAGE_KEY); + } catch (e) { + /* ignore */ + } + window.location.href = "/static/profile.html"; + } catch (err) { + el.loading.classList.remove("active"); + el.flow.style.display = "block"; + el.flow.innerHTML = + `
Re-run failed.
` + + `${err.message}

Your answers are still saved.
` + `
` + `
`; } } - // Kick off a conversation as soon as the page loads so the id is ready. - startConversation().catch(() => { + // Resume an in-progress conversation if one exists, otherwise start a + // fresh one so the id is ready by submit time. + render(); + resumeOrStart().catch(() => { /* will retry on submit */ }); - render(); diff --git a/scripts/reprocess_csv.py b/scripts/reprocess_csv.py new file mode 100644 index 0000000..99d047b --- /dev/null +++ b/scripts/reprocess_csv.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +"""Re-run AI extraction over a saved discovery CSV, offline. + +Reads a per-conversation CSV (``data/questions/{id}.csv``) or a row-slice of +the master log (``data/questions_master.csv`` filtered by ``--conversation-id``) +and runs the same DiscoveryExtractor the web app uses. By default it prints the +resulting profile JSON to stdout so it is safe to run for inspection; pass +``--write-db`` to also persist a new DiscoveryProfile. + +Examples: + python scripts/reprocess_csv.py data/questions/.csv + python scripts/reprocess_csv.py data/questions_master.csv \\ + --conversation-id --write-db + +Requires ANTHROPIC_API_KEY (and optionally ANTHROPIC_MODEL) in the environment. +""" +import argparse +import asyncio +import csv +import json +import os +import sys +import uuid +from datetime import datetime, timezone + +# Allow running as a plain script (python scripts/reprocess_csv.py ...). +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from app.services.answer_store import DISCOVERY_PROMPTS # noqa: E402 +from app.services.extractor import ( # noqa: E402 + DiscoveryExtractionError, + DiscoveryExtractor, +) + +PROMPT_KEYS = [p["key"] for p in DISCOVERY_PROMPTS] + + +def read_answers(path: str, conversation_id: str | None) -> tuple[str, dict]: + """Return (conversation_id, {prompt_key: answer}) from a CSV file. + + If ``conversation_id`` is given, only rows for that conversation are used + (needed for the master log). Later rows win, so the most recent saved + answers take precedence. + """ + answers: dict[str, str] = {} + found_id = conversation_id + with open(path, newline="", encoding="utf-8") as fh: + for row in csv.DictReader(fh): + row_id = row.get("conversation_id") + if conversation_id and row_id != conversation_id: + continue + found_id = found_id or row_id + key = row.get("prompt_key") + if key in PROMPT_KEYS: + answers[key] = row.get("answer", "") or "" + if not answers: + raise SystemExit( + f"No matching answer rows found in {path}" + + (f" for conversation {conversation_id}" if conversation_id else "") + ) + return found_id or "", answers + + +async def run_extraction(answers: dict) -> dict: + api_key = os.getenv("ANTHROPIC_API_KEY") + model = os.getenv("ANTHROPIC_MODEL", "claude-sonnet-4-6") + responses = {k: answers.get(k, "") for k in PROMPT_KEYS} + extractor = DiscoveryExtractor(api_key=api_key, model=model) + return await extractor.extract(responses) + + +async def write_db(conversation_id: str, data: dict) -> str: + """Persist a new DiscoveryProfile for an existing conversation.""" + from app.database import AsyncSessionLocal + from app.models import DiscoveryConversation, DiscoveryProfile + + def _as_int(value): + try: + return int(value) + except (TypeError, ValueError): + return None + + async with AsyncSessionLocal() as db: + conversation = await db.get(DiscoveryConversation, conversation_id) + if conversation is None: + raise SystemExit( + f"Conversation {conversation_id} not found in the database; " + "cannot --write-db." + ) + profile = DiscoveryProfile( + id=str(uuid.uuid4()), + user_id=conversation.user_id, + conversation_id=conversation.id, + generated_at=datetime.now(timezone.utc), + triad=data.get("triad"), + probable_type=_as_int(data.get("probable_type")), + wing=_as_int(data.get("wing")), + instinctual_variant=data.get("instinctual_variant"), + instinctual_stack=data.get("instinctual_stack"), + love_summary=data.get("love_summary"), + strength_summary=data.get("strength_summary"), + mission_summary=data.get("mission_summary"), + vocation_summary=data.get("vocation_summary"), + overlap_narrative=data.get("overlap_narrative"), + short_term_goals=data.get("short_term_goals"), + long_term_goals=data.get("long_term_goals"), + confidence_json=json.dumps(data.get("confidence", {})), + locked=False, + ) + db.add(profile) + await db.commit() + return profile.id + + +async def main_async(args: argparse.Namespace) -> int: + conversation_id, answers = read_answers(args.csv_path, args.conversation_id) + try: + data = await run_extraction(answers) + except DiscoveryExtractionError as exc: + print(f"Extraction failed: {exc}", file=sys.stderr) + return 1 + + print(json.dumps(data, indent=2, ensure_ascii=False)) + + if args.write_db: + if not conversation_id: + print( + "Cannot --write-db: no conversation_id in the CSV.", + file=sys.stderr, + ) + return 1 + profile_id = await write_db(conversation_id, data) + print( + f"\nWrote profile {profile_id} for conversation {conversation_id}.", + file=sys.stderr, + ) + return 0 + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "csv_path", help="Path to a per-conversation or master CSV file." + ) + parser.add_argument( + "--conversation-id", + help="Only use rows for this conversation (required for the master log " + "when it holds more than one conversation).", + ) + parser.add_argument( + "--write-db", + action="store_true", + help="Persist a new DiscoveryProfile to the database (default: print " + "only).", + ) + args = parser.parse_args() + return asyncio.run(main_async(args)) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_answer_store.py b/tests/test_answer_store.py new file mode 100644 index 0000000..ac95627 --- /dev/null +++ b/tests/test_answer_store.py @@ -0,0 +1,117 @@ +"""Unit tests for the CSV answer store. + +These exercise the file I/O directly with a lightweight stand-in for the +DiscoveryConversation/User ORM objects, so they need no database. The +QUESTIONS_DIR / QUESTIONS_MASTER_CSV env vars are pointed at tmp_path. +""" +import csv +import os +from types import SimpleNamespace + +from app.services import answer_store + + +def _conversation(conv_id="conv-1", user_id="user-1", **answers): + base = { + "id": conv_id, + "user_id": user_id, + "prompt_alive": "", + "prompt_friction": "", + "prompt_pull": "", + "prompt_recognition": "", + "prompt_future": "", + "prompt_goals_short": "", + "prompt_goals_long": "", + } + base.update(answers) + return SimpleNamespace(**base) + + +def _point_env(monkeypatch, tmp_path): + qdir = tmp_path / "questions" + master = tmp_path / "questions_master.csv" + monkeypatch.setenv("QUESTIONS_DIR", str(qdir)) + monkeypatch.setenv("QUESTIONS_MASTER_CSV", str(master)) + return qdir, master + + +def test_write_conversation_csv_round_trips(monkeypatch, tmp_path): + _point_env(monkeypatch, tmp_path) + conv = _conversation(prompt_alive="felt alive", prompt_goals_long="big plans") + user = SimpleNamespace(email="a@b.com") + + assert answer_store.write_conversation_csv(conv, user, "responses_saved") + + path = answer_store.conversation_csv_path("conv-1") + assert os.path.exists(path) + + answers = answer_store.read_conversation_answers("conv-1") + assert answers["alive"] == "felt alive" + assert answers["goals_long"] == "big plans" + # Unanswered prompts round-trip as empty strings, all 7 present. + assert len(answers) == 7 + assert answers["friction"] == "" + + +def test_write_is_atomic_overwrite(monkeypatch, tmp_path): + _point_env(monkeypatch, tmp_path) + user = SimpleNamespace(email="a@b.com") + + answer_store.write_conversation_csv( + _conversation(prompt_alive="first"), user, "responses_saved" + ) + answer_store.write_conversation_csv( + _conversation(prompt_alive="second"), user, "completed" + ) + + answers = answer_store.read_conversation_answers("conv-1") + assert answers["alive"] == "second" + # No leftover temp file. + assert not os.path.exists( + answer_store.conversation_csv_path("conv-1") + ".tmp" + ) + + +def test_append_master_writes_header_once(monkeypatch, tmp_path): + _, master = _point_env(monkeypatch, tmp_path) + user = SimpleNamespace(email="a@b.com") + + answer_store.append_master( + _conversation("c1", prompt_alive="x"), user, "responses_saved" + ) + answer_store.append_master( + _conversation("c2", prompt_alive="y"), user, "completed" + ) + + with open(master, newline="", encoding="utf-8") as fh: + rows = list(csv.DictReader(fh)) + # 7 prompts per event, two events. + assert len(rows) == 14 + conv_ids = {r["conversation_id"] for r in rows} + assert conv_ids == {"c1", "c2"} + statuses = {r["status"] for r in rows} + assert statuses == {"responses_saved", "completed"} + + +def test_read_missing_file_returns_empty(monkeypatch, tmp_path): + _point_env(monkeypatch, tmp_path) + assert answer_store.read_conversation_answers("nope") == {} + + +def test_conversation_csv_text_from_db(monkeypatch, tmp_path): + _point_env(monkeypatch, tmp_path) + conv = _conversation(prompt_pull="natural pull") + user = SimpleNamespace(email="a@b.com") + text = answer_store.conversation_csv_text(conv, user, "completed") + assert "prompt_key" in text # header present + assert "natural pull" in text + assert text.count("\n") >= 8 # header + 7 rows + + +def test_save_answers_writes_both_artifacts(monkeypatch, tmp_path): + qdir, master = _point_env(monkeypatch, tmp_path) + conv = _conversation(prompt_alive="hi") + user = SimpleNamespace(email="a@b.com") + answer_store.save_answers(conv, user, "responses_saved") + assert os.path.exists(answer_store.conversation_csv_path("conv-1")) + assert os.path.exists(master) diff --git a/tests/test_discovery_csv.py b/tests/test_discovery_csv.py new file mode 100644 index 0000000..1087eaa --- /dev/null +++ b/tests/test_discovery_csv.py @@ -0,0 +1,158 @@ +"""Tests for the discovery CSV/reprocess endpoints. + +Run under the X-API-Key admin identity so they don't depend on the Google +OAuth flow. The DiscoveryExtractor is replaced with a fake so no network call +or API key is needed. QUESTIONS_DIR / QUESTIONS_MASTER_CSV are pointed at +tmp_path so the suite never writes into the repo's data dir. +""" +import os + +from app.services import answer_store + +API_KEY = {"X-API-Key": "test-api-key"} + + +class FakeExtractor: + """Stand-in for DiscoveryExtractor: returns a fixed well-formed profile.""" + + def __init__(self, api_key=None, model=None): + pass + + async def extract(self, responses): + return { + "triad": "gut", + "probable_type": 8, + "wing": 9, + "instinctual_variant": "sp", + "instinctual_stack": "sp/so/sx", + "love_summary": "love", + "strength_summary": "strength", + "mission_summary": "mission", + "vocation_summary": "vocation", + "overlap_narrative": "narrative", + "short_term_goals": "short", + "long_term_goals": "long", + "confidence": { + "triad": "high", + "type": "high", + "variant": "medium", + "ikigai": "high", + }, + "extraction_notes": "ok", + } + + +def _patch(monkeypatch, tmp_path): + monkeypatch.setenv("QUESTIONS_DIR", str(tmp_path / "questions")) + monkeypatch.setenv("QUESTIONS_MASTER_CSV", str(tmp_path / "master.csv")) + monkeypatch.setattr( + "app.routers.discovery.DiscoveryExtractor", FakeExtractor + ) + + +async def _start_and_respond(app_client): + start = await app_client.post("/discovery/start", headers=API_KEY) + assert start.status_code == 200 + conv_id = start.json()["conversation_id"] + r = await app_client.put( + f"/discovery/{conv_id}/respond", + headers=API_KEY, + json={ + "prompt_alive": "I felt alive building things", + "prompt_friction": "unfairness bothers me", + "prompt_goals_short": "ship the app", + }, + ) + assert r.status_code == 200 + return conv_id + + +async def test_respond_writes_csv(app_client, monkeypatch, tmp_path): + _patch(monkeypatch, tmp_path) + conv_id = await _start_and_respond(app_client) + + assert os.path.exists(answer_store.conversation_csv_path(conv_id)) + answers = answer_store.read_conversation_answers(conv_id) + assert answers["alive"] == "I felt alive building things" + assert os.path.exists(str(tmp_path / "master.csv")) + + +async def test_get_answers(app_client, monkeypatch, tmp_path): + _patch(monkeypatch, tmp_path) + conv_id = await _start_and_respond(app_client) + + r = await app_client.get(f"/discovery/{conv_id}/answers", headers=API_KEY) + assert r.status_code == 200 + body = r.json() + assert body["conversation_id"] == conv_id + assert body["completed_at"] is None + by_key = {a["prompt_key"]: a["answer"] for a in body["answers"]} + assert by_key["friction"] == "unfairness bothers me" + assert len(body["answers"]) == 7 + + +async def test_download_answers_csv(app_client, monkeypatch, tmp_path): + _patch(monkeypatch, tmp_path) + conv_id = await _start_and_respond(app_client) + + r = await app_client.get( + f"/discovery/{conv_id}/answers.csv", headers=API_KEY + ) + assert r.status_code == 200 + assert r.headers["content-type"].startswith("text/csv") + assert "attachment" in r.headers["content-disposition"] + assert "ship the app" in r.text + + +async def test_reprocess_creates_profile(app_client, monkeypatch, tmp_path): + _patch(monkeypatch, tmp_path) + conv_id = await _start_and_respond(app_client) + + r = await app_client.post( + f"/discovery/{conv_id}/reprocess", headers=API_KEY + ) + assert r.status_code == 200 + body = r.json() + assert body["conversation_id"] == conv_id + assert body["triad"] == "gut" + assert body["probable_type"] == 8 + + # A profile now exists for the user. + prof = await app_client.get("/discovery/profile/me", headers=API_KEY) + assert prof.status_code == 200 + + +async def test_reprocess_locked_returns_409(app_client, monkeypatch, tmp_path): + _patch(monkeypatch, tmp_path) + conv_id = await _start_and_respond(app_client) + + # Complete to create a profile, then lock it via confirm. + complete = await app_client.post( + f"/discovery/{conv_id}/complete", headers=API_KEY + ) + assert complete.status_code == 200 + confirm = await app_client.put( + "/discovery/profile/me/confirm", headers=API_KEY + ) + assert confirm.status_code == 200 + + r = await app_client.post( + f"/discovery/{conv_id}/reprocess", headers=API_KEY + ) + assert r.status_code == 409 + + +async def test_complete_writes_completed_csv(app_client, monkeypatch, tmp_path): + _patch(monkeypatch, tmp_path) + conv_id = await _start_and_respond(app_client) + + complete = await app_client.post( + f"/discovery/{conv_id}/complete", headers=API_KEY + ) + assert complete.status_code == 200 + + # The answers endpoint now reports a completion timestamp. + answers = await app_client.get( + f"/discovery/{conv_id}/answers", headers=API_KEY + ) + assert answers.json()["completed_at"] is not None