Persist discovery answers to CSV for reprocessing and recovery

Save each discovery conversation's prompts and answers to durable CSV
files (per-conversation + append-only master log) on both save and
completion, so answers survive an extraction error, can be re-fed to the
AI, and can be reviewed/resumed by the user.

- app/services/answer_store.py: canonical prompt list + atomic CSV writes,
  master append, and read-back helpers (DB stays system of record; CSV
  failures are logged, never fatal).
- discovery router: write CSV on /respond and /complete; new endpoints
  GET /answers, GET /answers.csv, POST /reprocess (shared extraction
  helper; locked profiles return 409).
- discovery.html: prefill/resume from saved answers after an error and a
  "Re-run analysis" button wired to /reprocess.
- scripts/reprocess_csv.py: offline CLI to re-run extraction from a CSV
  (print or --write-db).
- QUESTIONS_DIR / QUESTIONS_MASTER_CSV config, .gitignore, README, tests.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Dg6XWUwprmP5QCL18HxssY
This commit is contained in:
Claude
2026-06-19 01:06:40 +00:00
parent c1daef6411
commit 866bd60225
10 changed files with 988 additions and 51 deletions
+6
View File
@@ -6,6 +6,12 @@ HOST_PORT=8011
# Optional: override the Anthropic model used for extraction # Optional: override the Anthropic model used for extraction
ANTHROPIC_MODEL=claude-sonnet-4-6 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 # Google OAuth (web client type). Register the redirect URI below as an
# authorized redirect URI in the Google Cloud Console for this client. # authorized redirect URI in the Google Cloud Console for this client.
GOOGLE_CLIENT_ID=xxxx.apps.googleusercontent.com GOOGLE_CLIENT_ID=xxxx.apps.googleusercontent.com
+2
View File
@@ -7,5 +7,7 @@ data/*.db
data/*.db-journal data/*.db-journal
data/*.log data/*.log
data/*.err.log data/*.err.log
data/questions/
data/questions_master.csv
.pytest_cache/ .pytest_cache/
*.egg-info/ *.egg-info/
+57
View File
@@ -196,6 +196,7 @@ Important files:
| `app/models.py` | SQLAlchemy ORM models for users, refresh tokens, activity log, conversations, and profiles | | `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/database.py` | Async database engine, session factory, SQLite directory setup |
| `app/services/extractor.py` | Anthropic client wrapper, prompt, JSON parsing, retry logic | | `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/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/discovery.html` | Browser-based seven-prompt flow |
| `app/static/profile.html` | Browser-based profile display, edit, and confirm actions; links to reflection | | `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/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/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 | | `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/001_initial.py` | Initial database schema migration |
| `alembic/versions/002_add_auth.py` | Adds `users`, `refresh_tokens`, and `activity_log` tables | | `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` | | `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_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_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_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_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_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 | | `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`. 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 ### 4. Completing Analysis
`POST /discovery/{conversation_id}/complete` loads the conversation, builds a `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 On success, it stores a new `DiscoveryProfile`, marks the conversation
`completed_at`, and returns the profile. `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/<conversation_id>.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 <conversation_id> --write-db
```
It reads `ANTHROPIC_API_KEY` (and optional `ANTHROPIC_MODEL`) from the
environment.
### 5. Loading The Profile ### 5. Loading The Profile
`GET /discovery/profile/me` fetches the newest profile for the authenticated `GET /discovery/profile/me` fetches the newest profile for the authenticated
+147 -46
View File
@@ -11,6 +11,7 @@ import uuid
from datetime import datetime, timezone from datetime import datetime, timezone
from fastapi import APIRouter, Depends, HTTPException from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import Response
from sqlalchemy import select from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
@@ -24,6 +25,8 @@ from app.models import (
ReflectionMessage, ReflectionMessage,
User, 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.extractor import DiscoveryExtractionError, DiscoveryExtractor
from app.services.reflector import ( from app.services.reflector import (
EDITABLE_FIELDS, 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( def _to_profile_response(
profile: DiscoveryProfile, extraction_notes: str | None = None profile: DiscoveryProfile, extraction_notes: str | None = None
) -> schemas.ProfileResponse: ) -> schemas.ProfileResponse:
@@ -161,6 +218,10 @@ async def save_responses(
conversation.prompt_goals_long = payload.prompt_goals_long conversation.prompt_goals_long = payload.prompt_goals_long
await db.commit() 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( return schemas.RespondResponse(
conversation_id=conversation_id, status="responses_saved" conversation_id=conversation_id, status="responses_saved"
) )
@@ -176,59 +237,99 @@ async def complete_conversation(
): ):
conversation = await _owned_conversation(db, conversation_id, user) conversation = await _owned_conversation(db, conversation_id, user)
responses = { profile, extraction_notes = await _generate_profile(
"alive": conversation.prompt_alive or "", db, conversation, source="extraction"
"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,
) )
conversation.completed_at = _now() conversation.completed_at = _now()
db.add(profile)
_record_revision(db, profile, source="extraction")
await db.commit() await db.commit()
return _to_profile_response( # Refresh the durable CSV copy now that the conversation is complete.
profile, extraction_notes=data.get("extraction_notes") 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) @router.get("/profile/me", response_model=schemas.ProfileResponse)
async def get_my_profile( async def get_my_profile(
db: AsyncSession = Depends(get_db), db: AsyncSession = Depends(get_db),
+15
View File
@@ -24,6 +24,21 @@ class RespondResponse(BaseModel):
status: str 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): class Confidence(BaseModel):
triad: Optional[str] = None triad: Optional[str] = None
type: Optional[str] = None type: Optional[str] = None
+223
View File
@@ -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 (612 months)"},
{"key": "goals_long", "column": "prompt_goals_long", "title": "Long-Term Goals (35 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
+102 -5
View File
@@ -133,11 +133,62 @@
} }
}); });
const STORAGE_KEY = "discovery_conversation_id";
async function startConversation() { async function startConversation() {
const res = await authedFetch("/discovery/start", { method: "POST" }); const res = await authedFetch("/discovery/start", { method: "POST" });
if (!res.ok) throw new Error("Could not start conversation"); if (!res.ok) throw new Error("Could not start conversation");
const data = await res.json(); const data = await res.json();
conversationId = data.conversation_id; 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() { async function submit() {
@@ -173,24 +224,70 @@
throw new Error(detail.detail || "Extraction failed"); 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"; window.location.href = "/static/profile.html";
} catch (err) { } catch (err) {
el.loading.classList.remove("active"); el.loading.classList.remove("active");
el.flow.style.display = "block"; el.flow.style.display = "block";
el.flow.innerHTML = el.flow.innerHTML =
`<div class="error-box"><strong>Something went wrong.</strong><br/>` + `<div class="error-box"><strong>Something went wrong.</strong><br/>` +
`${err.message}<br/><br/>Your answers are still here — ` + `${err.message}<br/><br/>Your answers are saved — you can ` +
`please try submitting again.</div>` + `re-run the analysis or reload to keep editing.</div>` +
`<div class="nav">` +
`<button class="btn-ghost" onclick="location.reload()">Reload</button>` +
`<button class="btn-primary" id="rerunBtn">Re-run analysis</button>` +
`</div>`;
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 =
`<div class="error-box"><strong>Re-run failed.</strong><br/>` +
`${err.message}<br/><br/>Your answers are still saved.</div>` +
`<div class="nav"><span></span>` + `<div class="nav"><span></span>` +
`<button class="btn-primary" onclick="location.reload()">Reload</button></div>`; `<button class="btn-primary" onclick="location.reload()">Reload</button></div>`;
} }
} }
// Kick off a conversation as soon as the page loads so the id is ready. // Resume an in-progress conversation if one exists, otherwise start a
startConversation().catch(() => { // fresh one so the id is ready by submit time.
render();
resumeOrStart().catch(() => {
/* will retry on submit */ /* will retry on submit */
}); });
render();
</script> </script>
</body> </body>
</html> </html>
+161
View File
@@ -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/<id>.csv
python scripts/reprocess_csv.py data/questions_master.csv \\
--conversation-id <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())
+117
View File
@@ -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)
+158
View File
@@ -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