Files
impactflow_discovery/app/services/profile_history.py
Joel Salmon b9d7b0e22b Phase 5: iteration & polish (visualizations, goal history, smart tagging)
Final roadmap phase. No DB migration — it reads data already captured.

- Goal-evolution history: GET /discovery/profile/me/goal-history derives a
  per-goal timeline from the profile_revision snapshots (pure aggregator in
  app/services/profile_history.py).
- Smart tagging: POST /discovery/integration/suggest-foundation suggests which
  foundation a task builds toward + rationale/confidence (FoundationTagger,
  app/services/tagging.py). Suggestion only; the person confirms by posting the
  task mapping.
- Deeper goal-refinement: the reflect loop accepts an optional focus ("goals")
  that steers the coach toward sharpening goals — still mirror, not compass.
- Visualizations: visuals.html renders an Ikigai Venn and an Enneagram diagram
  (plain-language callouts, not the raw type number) plus the goal-evolution
  timeline; linked from profile.html.

Tests: 99 passing (added pure goal-history tests, goal-history + suggest
endpoint tests, reflect-focus passthrough; run in-container). README updated.

This completes the ImpactFlow Vision roadmap (Phases 1-5) on the Discovery side.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-16 21:37:33 -05:00

51 lines
1.6 KiB
Python

"""Phase 5: derive how a person's goals evolved from the profile_revision
snapshots already captured since Phase 2.
Pure functions over revision rows — no DB access, fully testable.
"""
from typing import Any, Dict, List
# The goal fields we track an evolution timeline for.
GOAL_FIELDS = ("short_term_goals", "long_term_goals")
def goal_history(revisions: List[Dict[str, Any]]) -> Dict[str, List[dict]]:
"""Build a per-goal timeline of distinct values over time.
Args:
revisions: each ``{"fields": {<field>: value, ...}, "source": str,
"created_at": datetime}``, in ascending chronological order.
Returns:
``{"short_term_goals": [{"value", "source", "at"}...],
"long_term_goals": [...]}`` — one entry per *change*; consecutive
identical values are collapsed so the timeline shows only when a goal
actually moved.
"""
timelines: Dict[str, List[dict]] = {f: [] for f in GOAL_FIELDS}
last: Dict[str, Any] = {f: _SENTINEL for f in GOAL_FIELDS}
for rev in revisions:
fields = rev.get("fields") or {}
for field in GOAL_FIELDS:
value = fields.get(field)
if value == last[field]:
continue
last[field] = value
timelines[field].append(
{
"value": value,
"source": rev.get("source"),
"at": rev.get("created_at"),
}
)
return timelines
class _Sentinel:
pass
# Distinct from None so a first snapshot whose goal is None still emits once.
_SENTINEL = _Sentinel()