Ir para o conteúdo

mindflow v0.3 — Plano de Desenvolvimento Sessão por Sessão

Tech Lead: use este documento como guia de execução. Cada sessão é autossuficiente dentro das dependências listadas. Convenção TDD: escreva o teste, rode e veja falhar (red), implemente (green), refatore se necessário.


S1 — Briefings Personalizados + Schema de Metadados v1

Objetivo: o endpoint /api/routine/briefing passa a aceitar template e intent no body, gerando briefings diferentes por intenção. Simultaneamente, todos os eventos novos são salvos com o payload normalizado (schema v1 da DT-001).

Dependências: nenhuma — pode começar aqui.

Arquivos a criar/modificar

Ação Arquivo
Modificar /workspace/mindflow/mindflow/routines/engine.pyrun_routine()intent do body
Modificar /workspace/mindflow/mindflow/api/server.pyapi_routine() repassa intent e template
Modificar /workspace/mindflow/mindflow/context/store.pylog_event() aplica schema v1 por default
Criar /workspace/mindflow/mindflow/routines/briefing_templates.py — prompts por intent
Criar /workspace/mindflow/tests/test_schema_v1.py
Criar /workspace/mindflow/tests/test_briefing_intent.py

Testes TDD (escrever ANTES do código)

tests/test_schema_v1.py

def test_log_event_inclui_campos_v1(store):
    store.log_event("web", "nota", {"text": "oi"})
    ev = store.recent_events(limit=1)[0]
    assert ev["payload"]["version"] == "1"
    assert ev["payload"]["source"] == "web"
    assert ev["payload"]["type"] == "capture"
    assert "importance" in ev["payload"]

def test_retrocompatibilidade_payload_antigo(store):
    # eventos sem version devem ser lidos sem erro
    store._raw_insert({"source": "web", "kind": "nota", "payload": {"text": "antigo"}})
    ev = store.recent_events(limit=1)[0]
    text = ev["payload"].get("text", "")
    assert text == "antigo"

tests/test_briefing_intent.py

def test_briefing_sem_intent_usa_default(client):
    r = client.post("/api/routine/briefing", json={})
    assert r.status_code == 200

def test_briefing_intent_manha(client):
    r = client.post("/api/routine/briefing", json={"intent": "manha"})
    assert r.status_code == 200

def test_briefing_intent_invalido_usa_default(client):
    r = client.post("/api/routine/briefing", json={"intent": "xyzfoo"})
    assert r.status_code == 200  # nunca 422, graceful fallback

Critério de aceite

  • POST /api/routine/briefing com {"intent": "manha"} retorna HTML diferente de {"intent": "noite"}
  • Todos os novos eventos no banco têm payload.version == "1", payload.source, payload.type
  • Eventos antigos são lidos sem erro (fallback retrocompatível)
  • pytest tests/test_schema_v1.py tests/test_briefing_intent.py — 100% verde

S2 — Ajudar com Brisa (rotina completa)

Objetivo: o card "Ajudar com brisa" (já renderizado no frontend) tem backend funcional. Usuário digita ideia vaga → IA estrutura em texto acionável + sugere próximo passo.

Dependências: S1 (schema v1 já disponível para salvar resultado).

Arquivos a criar/modificar

Ação Arquivo
Criar /workspace/mindflow/mindflow/routines/brisa_routine.py — prompt + lógica
Modificar /workspace/mindflow/mindflow/routines/engine.py — registrar "brisa" como routine_id válido
Modificar /workspace/mindflow/mindflow/api/server.pyapi_routine() aceita body com context para brisa
Criar /workspace/mindflow/tests/test_routine_brisa.py

Testes TDD

tests/test_routine_brisa.py

def test_brisa_retorna_html(client, monkeypatch):
    monkeypatch.setattr("mindflow.routines.brisa_routine.call_llm", lambda *a, **k: "Estruturei sua ideia.")
    r = client.post("/api/routine/brisa", json={"context": "quero melhorar minha saúde mas não sei como"})
    assert r.status_code == 200
    assert "<" in r.text  # é HTML

def test_brisa_sem_context_retorna_prompt_inicial(client, monkeypatch):
    monkeypatch.setattr("mindflow.routines.brisa_routine.call_llm", lambda *a, **k: "Me conta mais.")
    r = client.post("/api/routine/brisa", json={})
    assert r.status_code == 200

def test_brisa_salva_evento_schema_v1(store, monkeypatch):
    monkeypatch.setattr("mindflow.routines.brisa_routine.call_llm", lambda *a, **k: "ok")
    from mindflow.routines.engine import run_routine
    run_routine("brisa", store, context="test")
    ev = store.recent_events(limit=1)[0]
    assert ev["payload"]["routine_id"] == "brisa"
    assert ev["payload"]["type"] == "routine_result"

Critério de aceite

  • Card "Ajudar com brisa" no frontend, ao clicar no botão, chama POST /api/routine/brisa, exibe resultado
  • Resultado é salvo no banco com routine_id: "brisa", type: "routine_result" (schema v1)
  • Histórico da rotina aparece abaixo do card (via loadRoutineHistory('brisa'))
  • pytest tests/test_routine_brisa.py — verde

S3 — Bota o Papo em Dia (rotina completa)

Objetivo: rotina de atualização de contexto — usuário narra o que aconteceu recentemente, Brisa resume e salva os pontos-chave no ContextStore.

Dependências: S1 (schema v1), S2 (padrão de rotina estabelecido).

Arquivos a criar/modificar

Ação Arquivo
Criar /workspace/mindflow/mindflow/routines/papo_routine.py — prompt de resumo + extração de pontos
Modificar /workspace/mindflow/mindflow/routines/engine.py — registrar "papo"
Criar /workspace/mindflow/tests/test_routine_papo.py

Testes TDD

tests/test_routine_papo.py

def test_papo_processa_narrativa(client, monkeypatch):
    monkeypatch.setattr("mindflow.routines.papo_routine.call_llm", lambda *a, **k: "Resumo: você fez X.")
    r = client.post("/api/routine/papo", json={"context": "hoje eu fiz academia e fui ao médico"})
    assert r.status_code == 200

def test_papo_salva_schema_v1(store, monkeypatch):
    monkeypatch.setattr("mindflow.routines.papo_routine.call_llm", lambda *a, **k: "ok")
    from mindflow.routines.engine import run_routine
    run_routine("papo", store, context="narrativa")
    ev = store.recent_events(limit=1)[0]
    assert ev["payload"]["routine_id"] == "papo"
    assert ev["payload"]["version"] == "1"

def test_papo_sem_context_retorna_instrucao(client, monkeypatch):
    monkeypatch.setattr("mindflow.routines.papo_routine.call_llm", lambda *a, **k: "Me conta o que rolou.")
    r = client.post("/api/routine/papo", json={})
    assert r.status_code == 200

Critério de aceite

  • Card "Bota o papo em dia" funcional: textarea → envia → exibe resumo da Brisa
  • Evento salvo com routine_id: "papo", pontos-chave extraíveis do payload
  • Histórico renderiza corretamente
  • pytest tests/test_routine_papo.py — verde

S4 — O Doutor (rotina completa)

Objetivo: rotina de check-in de bem-estar — perguntas curtas, usuário responde, Brisa gera diagnóstico leve e sugere ação concreta para o dia.

Dependências: S1, S2, S3 (padrão consolidado de rotina).

Arquivos a criar/modificar

Ação Arquivo
Criar /workspace/mindflow/mindflow/routines/doutor_routine.py — prompt de check-in + diagnóstico
Modificar /workspace/mindflow/mindflow/routines/engine.py — registrar "doutor"
Criar /workspace/mindflow/tests/test_routine_doutor.py

Testes TDD

tests/test_routine_doutor.py

def test_doutor_retorna_diagnostico(client, monkeypatch):
    monkeypatch.setattr("mindflow.routines.doutor_routine.call_llm", lambda *a, **k: "Você parece cansado. Sugestão: pausa de 10 min.")
    r = client.post("/api/routine/doutor", json={"context": "estou bem, dormindo pouco"})
    assert r.status_code == 200

def test_doutor_importancia_marcante_quando_alerta(store, monkeypatch):
    # Se LLM detectar sinal de alerta, importance deve ser >= 2
    monkeypatch.setattr("mindflow.routines.doutor_routine.call_llm", lambda *a, **k: "ALERTA: estresse alto.")
    from mindflow.routines.engine import run_routine
    run_routine("doutor", store, context="estou mal, sem dormir há 3 dias")
    ev = store.recent_events(limit=1)[0]
    assert ev["payload"]["importance"] >= 2

def test_doutor_schema_v1(store, monkeypatch):
    monkeypatch.setattr("mindflow.routines.doutor_routine.call_llm", lambda *a, **k: "ok")
    from mindflow.routines.engine import run_routine
    run_routine("doutor", store, context="tudo bem")
    ev = store.recent_events(limit=1)[0]
    assert ev["payload"]["routine_id"] == "doutor"

Critério de aceite

  • Card "O Doutor" funcional: usuário descreve como está → recebe diagnóstico + sugestão
  • Eventos com importance: 2 quando há sinal de atenção no relato
  • As 4 rotinas (briefing, brisa, papo, doutor) funcionam sem regressão
  • pytest tests/test_routine_doutor.py — verde

S5 — Copa do Mundo (ESPN API) + Aba Eventos

Objetivo: nova aba "Eventos" no frontend com resultados e próximos jogos via ESPN API. Endpoint /api/events/copa retorna dados estruturados.

Dependências: nenhuma das rotinas (independente). Pode rodar em paralelo com S2-S4 se necessário.

Arquivos a criar/modificar

Ação Arquivo
Criar /workspace/mindflow/mindflow/api/events.py — router FastAPI para /api/events/*
Criar /workspace/mindflow/mindflow/integrations/espn.py — client ESPN API
Modificar /workspace/mindflow/mindflow/api/server.py — incluir events.router
Modificar /workspace/mindflow/mindflow/frontend/templates/index.html — adicionar aba "Eventos"
Modificar /workspace/mindflow/mindflow/frontend/templates/app.jsloadCopa(), renderização
Criar /workspace/mindflow/tests/test_events_copa.py

Testes TDD

tests/test_events_copa.py

def test_copa_retorna_estrutura_correta(client, monkeypatch):
    monkeypatch.setattr("mindflow.integrations.espn.fetch_copa_data", lambda: {
        "jogos": [{"time_a": "Brasil", "time_b": "Argentina", "placar": "2-1", "status": "encerrado"}],
        "proximos": [{"time_a": "Brasil", "time_b": "Alemanha", "data": "2026-07-10"}]
    })
    r = client.get("/api/events/copa")
    assert r.status_code == 200
    data = r.json()
    assert "jogos" in data
    assert "proximos" in data

def test_copa_espn_indisponivel_retorna_fallback(client, monkeypatch):
    monkeypatch.setattr("mindflow.integrations.espn.fetch_copa_data", lambda: (_ for _ in ()).throw(Exception("timeout")))
    r = client.get("/api/events/copa")
    assert r.status_code == 200
    assert r.json().get("ok") is False or "error" in r.json()

def test_copa_cache_nao_chama_api_duas_vezes(monkeypatch):
    calls = []
    def mock_fetch():
        calls.append(1)
        return {"jogos": [], "proximos": []}
    monkeypatch.setattr("mindflow.integrations.espn.fetch_copa_data", mock_fetch)
    from mindflow.integrations.espn import get_copa_cached
    get_copa_cached()
    get_copa_cached()
    assert len(calls) == 1  # cache funcionou

Critério de aceite

  • GET /api/events/copa retorna {"jogos": [...], "proximos": [...]} com dados reais da ESPN
  • Aba "Eventos" visível no frontend, renderiza jogos e próximos
  • Se ESPN cair, endpoint retorna erro estruturado (não 500)
  • Cache de no mínimo 60s para não sobrecarregar a API externa
  • pytest tests/test_events_copa.py — verde

S6 — Bot Telegram com tool_use (3 ferramentas)

Objetivo: o bot Telegram do mindflow passa a usar tool_use com 3 ferramentas: get_context, set_context, create_note.

Dependências: S1 (schema v1 para create_note salvar corretamente).

Arquivos a criar/modificar

Ação Arquivo
Criar /workspace/mindflow/mindflow/bot/tools.py — definição das 3 ferramentas (schema JSON)
Criar /workspace/mindflow/mindflow/bot/tool_executor.py — executa tool call retornado pelo LLM
Modificar /workspace/mindflow/mindflow/bot/handler.py (ou equivalente) — loop tool_use
Criar /workspace/mindflow/tests/test_bot_tools.py

Testes TDD

tests/test_bot_tools.py

def test_get_context_retorna_snippets(store):
    store.log_event("telegram", "nota", {"text": "fui correr hoje"})
    from mindflow.bot.tool_executor import execute_tool
    result = execute_tool("get_context", {"query": "corrida"}, store=store, vstore=None)
    assert isinstance(result, str)

def test_set_context_salva_evento(store):
    from mindflow.bot.tool_executor import execute_tool
    execute_tool("set_context", {"key": "humor", "value": "animado"}, store=store, vstore=None)
    ev = store.recent_events(limit=1)[0]
    assert "animado" in ev["payload"]["text"]

def test_create_note_schema_v1(store):
    from mindflow.bot.tool_executor import execute_tool
    execute_tool("create_note", {"text": "ideia nova", "kind": "ideia"}, store=store, vstore=None)
    ev = store.recent_events(limit=1)[0]
    assert ev["payload"]["source"] == "telegram"
    assert ev["payload"]["version"] == "1"

def test_tool_desconhecida_nao_crasha(store):
    from mindflow.bot.tool_executor import execute_tool
    result = execute_tool("ferramenta_inexistente", {}, store=store, vstore=None)
    assert result is not None  # retorna mensagem de erro, não exceção

Critério de aceite

  • Mandar mensagem pro bot → ele usa tool_use quando pertinente (ex: "salva que eu fui correr")
  • create_note via bot salva com source: "telegram", schema v1
  • get_context busca no VectorStore se disponível, fallback no ContextStore
  • Bot nunca crasha por tool call inválida
  • pytest tests/test_bot_tools.py — verde

S7 — Arquivamento + Guardrail Soft de Setup

Objetivo: eventos podem ser arquivados (removidos do fluxo principal sem deletar). Guardrail verifica se setup foi feito e exibe aviso leve se não foi.

Dependências: S1 (schema v1), S3 (papo em dia para ter eventos a arquivar).

Arquivos a criar/modificar

Ação Arquivo
Modificar /workspace/mindflow/mindflow/context/store.pyarchive_event(event_id)
Modificar /workspace/mindflow/mindflow/api/server.pyPOST /api/event/{id}/archive e GET /api/context/setup-status
Modificar /workspace/mindflow/mindflow/frontend/templates/app.js — botão arquivar no timeline, banner setup
Criar /workspace/mindflow/tests/test_archive.py
Criar /workspace/mindflow/tests/test_setup_status.py

Testes TDD

tests/test_archive.py

def test_arquivar_evento_remove_do_timeline(client, store):
    store.log_event("web", "nota", {"text": "nota pra arquivar"})
    ev_id = store.recent_events(limit=1)[0]["id"]
    r = client.post(f"/api/event/{ev_id}/archive")
    assert r.json()["ok"] is True
    timeline = client.get("/api/timeline?min_importance=0&limit=50").json()
    ids = [i.get("id") for i in timeline["items"]]
    assert ev_id not in ids

def test_arquivar_evento_inexistente_retorna_404(client):
    r = client.post("/api/event/999999/archive")
    assert r.status_code == 404

tests/test_setup_status.py

def test_setup_status_falso_sem_setup(client):
    r = client.get("/api/context/setup-status")
    assert r.json()["configured"] is False

def test_setup_status_verdadeiro_apos_setup(client, store):
    store.log_event("web", "setup", {"text": "Perfil configurado", "type": "system"})
    r = client.get("/api/context/setup-status")
    assert r.json()["configured"] is True

Critério de aceite

  • Botão "Arquivar" no timeline funciona: evento some do feed, mas não é deletado do banco
  • GET /api/context/setup-status retorna {"configured": false} na primeira vez
  • Banner/aviso leve aparece no frontend se configured: false (não bloqueia uso)
  • pytest tests/test_archive.py tests/test_setup_status.py — verde

S8 — CI/CD + Testes Finais

Objetivo: pipeline GitHub Actions roda pytest em todo push para mindflow/main. Falha bloqueia merge. Todos os testes das sessões anteriores passam juntos.

Dependências: S1–S7 completos.

Arquivos a criar/modificar

Ação Arquivo
Criar /workspace/mindflow/.github/workflows/ci.yml
Modificar /workspace/mindflow/pyproject.toml ou setup.cfg — garantir que pytest está configurado
Criar /workspace/mindflow/tests/test_integration_v03.py — smoke test end-to-end

CI/CD — ci.yml

name: mindflow CI
on:
  push:
    branches: [main]
  pull_request:
    branches: [main]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install -e ".[dev]"
        working-directory: mindflow
      - run: pytest tests/ -v --tb=short
        working-directory: mindflow

Teste de integração final

tests/test_integration_v03.py

def test_smoke_todas_rotinas(client, monkeypatch):
    """Todas as 4 rotinas respondem 200 sem LLM real."""
    for routine in ["briefing", "brisa", "papo", "doutor"]:
        monkeypatch.setattr("mindflow.routines.engine.call_llm", lambda *a, **k: "ok")
        r = client.post(f"/api/routine/{routine}", json={"context": "test"})
        assert r.status_code == 200, f"Rotina {routine} falhou"

def test_smoke_endpoints_v03(client):
    """Endpoints novos respondem sem crash."""
    assert client.get("/api/events/copa").status_code in (200, 503)
    assert client.get("/api/context/setup-status").status_code == 200
    assert client.get("/api/timeline").status_code == 200

Critério de aceite

  • pytest tests/ -v — 100% verde localmente
  • Push para main dispara workflow e passa
  • PR com teste falhando é bloqueado pelo branch protection
  • CHANGELOG atualizado com todas as features da v0.3

Resumo das sessões

Sessão Escopo TDD Dependências
S1 Briefings Personalizados + Schema v1 6 testes nenhuma
S2 Ajudar com Brisa 3 testes S1
S3 Bota o Papo em Dia 3 testes S1, S2
S4 O Doutor 3 testes S1, S2, S3
S5 Copa do Mundo + Aba Eventos 3 testes independente
S6 Bot com tool_use 4 testes S1
S7 Arquivamento + Guardrail Setup 4 testes S1, S3
S8 CI/CD + Testes Finais 2 testes S1–S7