Fase 81.0 — Importador `organized-app` — Plan de Implementación

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Crear el package jw-meeting-scheduler con un importador que convierte un backup JSON de sws2apps/organized-app en el store local cifrado del scheduler (PersonRecord SQLite + AssignmentHistoryEntry SQLite), con dry-run + diff + idempotencia + respeto CRDT.

Architecture: Package nuevo packages/jw-meeting-scheduler/ con tres responsabilidades aisladas: (1) models/ Pydantic v2 que reflejan los nuevos tipos del scheduler; (2) importer/ parser de backup organized-app y mappers a los modelos del scheduler; (3) store/ SQLite + FieldEncryptor para persistir. Sin solver aún (F81.3), sin agente (F81.4) — esta fase pura import + persist.

Tech Stack: Python 3.13 · uv workspace · Pydantic v2 · SQLite (stdlib) · jw_core.privacy.encryption.FieldEncryptor · jw_core.models_organized (consumido, no modificado) · cryptography>=42 (transitive vía jw-core) · Typer + Rich para CLI · pytest + pytest-recording (no requiere red en esta fase).

Global Constraints

Python >=3.13 (requires-python uniforme con monorepo).
GPL-3.0-only licence header en todos los archivos nuevos del package.
Ruff lint + format: line-length 120, target-version = "py313", mismas reglas que pyproject.toml raíz.
Mypy strict: nuevo package se añade a [tool.mypy] config si necesario.
Cero red en tests: ningún test toca HTTP. Fixtures locales son únicos.
Cero LLM en este package: importer es 100% determinístico.
FieldEncryptor no instanciar Fernet directo: usar from jw_core.privacy.encryption import FieldEncryptor. Output cifrado es str (base64 token), nunca bytes.
Imports relativos prohibidos: usar from jw_meeting_scheduler.X import Y siempre.
DRY · YAGNI · TDD · commits frecuentes: cada task acaba en commit.
Multi-congregación: aislamiento por carpeta ~/.jw-agent-toolkit/congregations/<congregation_id>/. congregation_id matchea ^[a-z0-9_-]{3,64}$.
CRDT-aware: nunca sobrescribir local.last_updated si imported < local.
No tocar el núcleo jw-core salvo para añadir el package al workspace.

Task 1: Scaffold del package `jw-meeting-scheduler`

Files:

Create: packages/jw-meeting-scheduler/pyproject.toml
Create: packages/jw-meeting-scheduler/src/jw_meeting_scheduler/__init__.py
Create: packages/jw-meeting-scheduler/src/jw_meeting_scheduler/py.typed
Create: packages/jw-meeting-scheduler/tests/__init__.py
Create: packages/jw-meeting-scheduler/tests/test_scaffold.py
Modify: pyproject.toml (root, sección [tool.uv.workspace] + [tool.uv.sources] + [tool.pytest.ini_options])

Interfaces:

Consumes: nothing
Produces: package jw_meeting_scheduler importable; __version__ = "0.1.0".
Step 1: Crear pyproject.toml del package

# packages/jw-meeting-scheduler/pyproject.toml
[project]
name = "jw-meeting-scheduler"
version = "0.1.0"
description = "CP-SAT solver + organized-app importer for JW congregation meeting assignments."
readme = "README.md"
requires-python = ">=3.13"
license = "GPL-3.0-only"
authors = [
    { name = "Elias", email = "elias@cipreholding.com" }
]
dependencies = [
    "pydantic>=2.9.0",
    "jw-core",
    "typer>=0.13.0",
    "rich>=13.9.0",
]

[project.optional-dependencies]
solver = ["ortools>=9.10.4067"]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["src/jw_meeting_scheduler"]

Step 2: Crear __init__.py mínimo

# packages/jw-meeting-scheduler/src/jw_meeting_scheduler/__init__.py
"""jw-meeting-scheduler — CP-SAT solver + organized-app importer.

See docs/superpowers/specs/2026-06-17-fase-81-meeting-scheduler-design.md
"""

__version__ = "0.1.0"

__all__ = ["__version__"]

Step 3: Crear py.typed marker (empty file)

Run: touch packages/jw-meeting-scheduler/src/jw_meeting_scheduler/py.typed

Step 4: README mínimo

<!-- packages/jw-meeting-scheduler/README.md -->
# jw-meeting-scheduler

CP-SAT solver + `organized-app` importer for JW congregation meeting assignments.

Spec: [`docs/superpowers/specs/2026-06-17-fase-81-meeting-scheduler-design.md`](../../docs/superpowers/specs/2026-06-17-fase-81-meeting-scheduler-design.md).

Step 5: Añadir el package al workspace raíz

Edit pyproject.toml (raíz):

[tool.uv.workspace]
members = [
    "packages/jw-core",
    "packages/jw-cli",
    "packages/jw-mcp",
    "packages/jw-rag",
    "packages/jw-agents",
    "packages/jw-finetune",
    "packages/jw-eval",
    "packages/jw-gen",
    "packages/jw-brain",
    "packages/jw-meeting-media",
    "packages/jw-meeting-scheduler",   # ← nuevo
    "packages/jw-interp",
    "packages/create-jw-agent",
    "tools/pytest-cookbook",
]

[tool.uv.sources]
jw-meeting-scheduler = { workspace = true }   # ← nuevo (orden alfabético)

[tool.pytest.ini_options]
testpaths = [
    "packages/jw-core/tests",
    "packages/jw-mcp/tests",
    "packages/jw-cli/tests",
    "packages/jw-rag/tests",
    "packages/jw-agents/tests",
    "packages/jw-eval/tests",
    "packages/jw-gen/tests",
    "packages/jw-meeting-scheduler/tests",  # ← nuevo
    "tests",
]

Step 6: Escribir el test de scaffold

# packages/jw-meeting-scheduler/tests/test_scaffold.py
"""Smoke test that the package imports cleanly."""

import jw_meeting_scheduler


def test_package_imports() -> None:
    assert jw_meeting_scheduler.__version__ == "0.1.0"

Step 7: Sincronizar workspace y correr el test

Run: uv sync --all-packages

Run: .venv/bin/python -m pytest packages/jw-meeting-scheduler/tests/test_scaffold.py -v

Expected: 1 passed.

Step 8: Commit

git add packages/jw-meeting-scheduler pyproject.toml uv.lock
git commit -m "feat(meeting-scheduler): scaffold package (F81.0 task 1)"

Task 2: Modelos Pydantic del scheduler

Files:

Create: packages/jw-meeting-scheduler/src/jw_meeting_scheduler/models.py
Create: packages/jw-meeting-scheduler/tests/test_models.py

Interfaces:

Consumes: jw_core.models_organized.assignment.AssignmentCode.
Produces: PersonRecord, TimeAway, AssignmentHistoryEntry, Gender, Privilege, Status, SkillLevel, ImportSource types.
Step 1: Write the failing test

# packages/jw-meeting-scheduler/tests/test_models.py
"""Pydantic models for scheduler store."""

from datetime import date

import pytest

from jw_core.models_organized.assignment import AssignmentCode
from jw_meeting_scheduler.models import (
    AssignmentHistoryEntry,
    PersonRecord,
    TimeAway,
)


def test_person_record_minimal_ok() -> None:
    record = PersonRecord(
        person_id="juan-perez",
        display_name_ciphered="Juan Pérez",
        gender="male",
        last_updated="2026-06-17T10:00:00",
    )
    assert record.status == "active"
    assert record.privileges == []
    assert record.eligible_assignments == []
    assert record.is_midweek_student is False
    assert record.imported_from == "manual"


def test_person_record_rejects_bad_id() -> None:
    with pytest.raises(ValueError):
        PersonRecord(
            person_id="JP!",  # uppercase + non-allowed char
            display_name_ciphered="Juan",
            gender="male",
            last_updated="2026-06-17T10:00:00",
        )


def test_time_away_dates_roundtrip() -> None:
    ta = TimeAway(start=date(2026, 7, 1), end=date(2026, 7, 15), reason="holiday")
    assert ta.start.year == 2026
    assert ta.end.month == 7


def test_assignment_history_entry_required_fields() -> None:
    entry = AssignmentHistoryEntry(
        entry_id="hist-001",
        person_id="juan-perez",
        assignment_field="MM_TGWBibleReading_A",
        assignment_code=AssignmentCode.MM_BibleReading,
        meeting_date=date(2026, 7, 6),
        updated_at="2026-06-17T10:00:00",
    )
    assert entry.aula == "main_hall"
    assert entry.confirmed is False
    assert entry.cancelled is False

Step 2: Run test (expect FAIL — models don’t exist)

Run: .venv/bin/python -m pytest packages/jw-meeting-scheduler/tests/test_models.py -v

Expected: ModuleNotFoundError: No module named 'jw_meeting_scheduler.models'.

Step 3: Write models

# packages/jw-meeting-scheduler/src/jw_meeting_scheduler/models.py
"""Pydantic models for the meeting-scheduler local store.

Mirrors organized-app data into a flatter, scheduler-friendly shape.
The importer (`jw_meeting_scheduler.importer`) is the only producer.
"""

from __future__ import annotations

from datetime import date
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field

from jw_core.models_organized.assignment import AssignmentCode

Gender = Literal["male", "female", "unknown"]
Privilege = Literal["elder", "ms"]
Status = Literal["active", "irregular", "inactive", "disfellowshipped", "deceased"]
SkillLevel = Literal[1, 2, 3, 4, 5]
ImportSource = Literal["organized_app", "manual"]
Aula = Literal["main_hall", "aux_class_1", "aux_class_2"]


class TimeAway(BaseModel):
    """One range during which the person is unavailable."""

    model_config = ConfigDict(populate_by_name=True)

    start: date
    end: date
    reason: str = ""


class PersonRecord(BaseModel):
    """Scheduler-flat publisher record. Encrypted display_name via FieldEncryptor."""

    model_config = ConfigDict(populate_by_name=True)

    person_id: str = Field(pattern=r"^[a-z0-9_-]{3,64}$")
    display_name_ciphered: str
    gender: Gender
    status: Status = "active"
    is_midweek_student: bool = False
    privileges: list[Privilege] = Field(default_factory=list)
    eligible_assignments: list[AssignmentCode] = Field(default_factory=list)
    skill_level: dict[AssignmentCode, SkillLevel] = Field(default_factory=dict)
    languages: list[str] = Field(default_factory=lambda: ["en"])
    time_away: list[TimeAway] = Field(default_factory=list)
    last_updated: str  # ISO-8601 — CRDT timestamp seed
    imported_from: ImportSource = "manual"


class AssignmentHistoryEntry(BaseModel):
    """One historical (or just-confirmed) assignment of a person to a slot."""

    model_config = ConfigDict(populate_by_name=True)

    entry_id: str
    person_id: str
    assignment_field: str           # e.g. "MM_AYFPart1_Student_A"
    assignment_code: AssignmentCode
    meeting_date: date
    aula: Aula = "main_hall"
    confirmed: bool = False
    confirmed_at: str | None = None
    cancelled: bool = False
    cancellation_reason: str = ""
    updated_at: str                  # ISO-8601 CRDT timestamp

Step 4: Run tests

Run: .venv/bin/python -m pytest packages/jw-meeting-scheduler/tests/test_models.py -v

Expected: 4 passed.

Step 5: Commit

git add packages/jw-meeting-scheduler/src/jw_meeting_scheduler/models.py packages/jw-meeting-scheduler/tests/test_models.py
git commit -m "feat(meeting-scheduler): Pydantic models PersonRecord + AssignmentHistoryEntry (F81.0 task 2)"

Task 3: Encryption helper sobre `FieldEncryptor`

Files:

Create: packages/jw-meeting-scheduler/src/jw_meeting_scheduler/crypto.py
Create: packages/jw-meeting-scheduler/tests/test_crypto.py

Interfaces:

Consumes: jw_core.privacy.encryption.FieldEncryptor, derive_key_from_password.
Produces: get_encryptor(passphrase: str | None, congregation_id: str) -> FieldEncryptor.
Step 1: Write the failing test

# packages/jw-meeting-scheduler/tests/test_crypto.py
"""Encryption helper wraps FieldEncryptor and derives a key per congregation."""

from __future__ import annotations

import os

from jw_meeting_scheduler.crypto import get_encryptor


def test_no_passphrase_no_env_returns_noop_encryptor(monkeypatch) -> None:
    monkeypatch.delenv("JW_PRIVACY_KEY", raising=False)
    enc = get_encryptor(passphrase=None, congregation_id="test-cong")
    assert enc.enabled is False
    # No-op: encrypt is identity-ish.
    assert enc.encrypt("hello") == "hello"


def test_passphrase_derives_key_and_round_trips(monkeypatch) -> None:
    monkeypatch.delenv("JW_PRIVACY_KEY", raising=False)
    enc = get_encryptor(passphrase="correct-horse-battery-staple", congregation_id="test-cong")
    assert enc.enabled is True
    cipher = enc.encrypt("Juan Pérez")
    assert cipher != "Juan Pérez"
    assert enc.decrypt(cipher) == "Juan Pérez"


def test_different_congregation_derives_different_key(monkeypatch) -> None:
    monkeypatch.delenv("JW_PRIVACY_KEY", raising=False)
    enc_a = get_encryptor(passphrase="same-passphrase", congregation_id="cong-a")
    enc_b = get_encryptor(passphrase="same-passphrase", congregation_id="cong-b")
    cipher_a = enc_a.encrypt("Juan Pérez")
    # Different salt → different key → enc_b cannot decrypt enc_a's output.
    try:
        decrypted = enc_b.decrypt(cipher_a)
        assert decrypted != "Juan Pérez"
    except Exception:
        pass  # expected: InvalidToken from Fernet

Step 2: Run test (expect FAIL — module doesn’t exist)

Run: .venv/bin/python -m pytest packages/jw-meeting-scheduler/tests/test_crypto.py -v

Expected: ModuleNotFoundError: No module named 'jw_meeting_scheduler.crypto'.

Step 3: Implement crypto helper

# packages/jw-meeting-scheduler/src/jw_meeting_scheduler/crypto.py
"""Encryption helper that wraps jw_core.privacy.encryption.FieldEncryptor.

Per-congregation salt makes the derived key distinct across congregations
even if the passphrase is shared between coordinators.
"""

from __future__ import annotations

from jw_core.privacy.encryption import FieldEncryptor, derive_key_from_password


def get_encryptor(passphrase: str | None, congregation_id: str) -> FieldEncryptor:
    """Return a FieldEncryptor for this congregation.

    - If `passphrase` is provided, derive a Fernet key via PBKDF2 with
      `congregation_id` as salt suffix.
    - Otherwise, FieldEncryptor reads JW_PRIVACY_KEY env var or falls back
      to no-op (identical to jw_core.ministry.field_report behaviour).
    """
    if passphrase:
        salt = b"jw-meeting-scheduler/v1:" + congregation_id.encode("utf-8")
        key = derive_key_from_password(passphrase, salt=salt)
        return FieldEncryptor(key=key)
    return FieldEncryptor()

Step 4: Run tests

Run: .venv/bin/python -m pytest packages/jw-meeting-scheduler/tests/test_crypto.py -v

Expected: 3 passed.

Step 5: Commit

git add packages/jw-meeting-scheduler/src/jw_meeting_scheduler/crypto.py packages/jw-meeting-scheduler/tests/test_crypto.py
git commit -m "feat(meeting-scheduler): encryption helper over FieldEncryptor with per-congregation salt (F81.0 task 3)"

Task 4: Loader del JSON backup `organized-app`

Files:

Create: packages/jw-meeting-scheduler/src/jw_meeting_scheduler/importer/__init__.py
Create: packages/jw-meeting-scheduler/src/jw_meeting_scheduler/importer/loader.py
Create: packages/jw-meeting-scheduler/tests/fixtures/__init__.py
Create: packages/jw-meeting-scheduler/tests/fixtures/backup_minimal.json
Create: packages/jw-meeting-scheduler/tests/test_importer_loader.py

Interfaces:

Consumes: nothing external (stdlib json + pathlib).
Produces: load_backup(path: Path) -> dict and OrganizedAppBackup Pydantic root model.
Step 1: Create the minimal fixture backup

{
  "schema_version": "3.0.0",
  "exported_at": "2026-06-15T10:00:00",
  "congregation": {
    "id": "kingdom-hall-test",
    "name": "Kingdom Hall Test",
    "languages": ["es"]
  },
  "persons": [
    {
      "_deleted": {"value": false, "updatedAt": "2026-01-01T00:00:00"},
      "person_uid": "uid-juan-perez",
      "person_data": {
        "person_firstname": {"value": "Juan", "updatedAt": "2026-01-01T00:00:00"},
        "person_lastname": {"value": "Pérez", "updatedAt": "2026-01-01T00:00:00"},
        "person_display_name": {"value": "Juan Pérez", "updatedAt": "2026-01-01T00:00:00"},
        "male": {"value": true, "updatedAt": "2026-01-01T00:00:00"},
        "female": {"value": false, "updatedAt": "2026-01-01T00:00:00"},
        "birth_date": {"value": "1985-03-15", "updatedAt": "2026-01-01T00:00:00"},
        "assignments": [
          {"type": "TGW_BibleReading", "updatedAt": "2026-01-01T00:00:00", "values": [100]},
          {"type": "AYF_Initial_Call", "updatedAt": "2026-01-01T00:00:00", "values": [101, 123]}
        ],
        "timeAway": [],
        "archived": {"value": false, "updatedAt": "2026-01-01T00:00:00"},
        "disqualified": {"value": false, "updatedAt": "2026-01-01T00:00:00"},
        "email": {"value": "", "updatedAt": "2026-01-01T00:00:00"},
        "address": {"value": "", "updatedAt": "2026-01-01T00:00:00"},
        "phone": {"value": "", "updatedAt": "2026-01-01T00:00:00"},
        "publisher_baptized": {
          "active": {"value": true, "updatedAt": "2026-01-01T00:00:00"},
          "anointed": {"value": false, "updatedAt": "2026-01-01T00:00:00"},
          "other_sheep": {"value": true, "updatedAt": "2026-01-01T00:00:00"},
          "baptism_date": {"value": "2005-06-12", "updatedAt": "2026-01-01T00:00:00"},
          "history": []
        },
        "publisher_unbaptized": {
          "active": {"value": false, "updatedAt": "2026-01-01T00:00:00"},
          "history": []
        },
        "midweek_meeting_student": {
          "active": {"value": true, "updatedAt": "2026-01-01T00:00:00"},
          "history": []
        },
        "privileges": [
          {
            "id": "priv-001",
            "_deleted": false,
            "updatedAt": "2026-01-01T00:00:00",
            "privilege": "ms",
            "start_date": "2020-01-15",
            "end_date": ""
          }
        ],
        "enrollments": [],
        "emergency_contacts": [],
        "family_members": {"head": true, "members": [], "updatedAt": "2026-01-01T00:00:00"}
      }
    }
  ],
  "schedules": []
}

Step 2: Write the failing test

# packages/jw-meeting-scheduler/tests/test_importer_loader.py
"""Loader reads the organized-app backup JSON into typed shape."""

from __future__ import annotations

from pathlib import Path

import pytest

from jw_meeting_scheduler.importer.loader import (
    BackupLoadError,
    load_backup,
)

FIXTURES = Path(__file__).parent / "fixtures"


def test_load_minimal_backup_succeeds() -> None:
    backup = load_backup(FIXTURES / "backup_minimal.json")
    assert backup.congregation.id == "kingdom-hall-test"
    assert backup.congregation.languages == ["es"]
    assert len(backup.persons) == 1
    p = backup.persons[0]
    assert p.person_uid == "uid-juan-perez"
    # PersonType nested envelope:
    assert p.person_data.person_display_name.value == "Juan Pérez"
    assert p.person_data.male.value is True
    assert p.person_data.female.value is False


def test_load_nonexistent_path_raises_BackupLoadError(tmp_path: Path) -> None:
    with pytest.raises(BackupLoadError, match="does not exist"):
        load_backup(tmp_path / "missing.json")


def test_load_malformed_json_raises_BackupLoadError(tmp_path: Path) -> None:
    bad = tmp_path / "bad.json"
    bad.write_text("{not json")
    with pytest.raises(BackupLoadError, match="invalid JSON"):
        load_backup(bad)


def test_load_missing_required_field_raises_BackupLoadError(tmp_path: Path) -> None:
    bad = tmp_path / "missing_field.json"
    bad.write_text('{"schema_version": "3.0.0"}')
    with pytest.raises(BackupLoadError, match="schema mismatch"):
        load_backup(bad)

Step 3: Run test (expect FAIL)

Run: .venv/bin/python -m pytest packages/jw-meeting-scheduler/tests/test_importer_loader.py -v

Expected: import error.

Step 4: Implement loader

# packages/jw-meeting-scheduler/src/jw_meeting_scheduler/importer/__init__.py
"""organized-app backup importer."""

from jw_meeting_scheduler.importer.loader import (
    BackupLoadError,
    OrganizedAppBackup,
    OrganizedAppCongregation,
    load_backup,
)

__all__ = [
    "BackupLoadError",
    "OrganizedAppBackup",
    "OrganizedAppCongregation",
    "load_backup",
]

# packages/jw-meeting-scheduler/src/jw_meeting_scheduler/importer/loader.py
"""Reads an organized-app backup JSON file into typed Pydantic shape.

The backup is the JSON exported from the organized-app web UI
(`Settings → Backup → Export`). Schema mirrors the TS definitions ported
in `jw_core.models_organized` plus the wrapping {congregation, persons,
schedules} envelope.
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

from pydantic import BaseModel, ConfigDict, ValidationError

from jw_core.models_organized.person import PersonType


class BackupLoadError(RuntimeError):
    """Raised when the backup file cannot be parsed or fails schema check."""


class OrganizedAppCongregation(BaseModel):
    model_config = ConfigDict(populate_by_name=True, extra="allow")
    id: str
    name: str
    languages: list[str] = []


class OrganizedAppBackup(BaseModel):
    model_config = ConfigDict(populate_by_name=True, extra="allow")
    schema_version: str
    exported_at: str
    congregation: OrganizedAppCongregation
    persons: list[PersonType] = []
    schedules: list[dict[str, Any]] = []   # typed in Task 6


def load_backup(path: Path) -> OrganizedAppBackup:
    if not path.exists():
        raise BackupLoadError(f"backup path does not exist: {path}")
    try:
        raw = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as e:
        raise BackupLoadError(f"invalid JSON in {path}: {e}") from e
    try:
        return OrganizedAppBackup.model_validate(raw)
    except ValidationError as e:
        raise BackupLoadError(f"schema mismatch in {path}: {e}") from e

Step 5: Run tests

Run: .venv/bin/python -m pytest packages/jw-meeting-scheduler/tests/test_importer_loader.py -v

Expected: 4 passed.

Step 6: Commit

git add packages/jw-meeting-scheduler/src/jw_meeting_scheduler/importer/ packages/jw-meeting-scheduler/tests/fixtures/ packages/jw-meeting-scheduler/tests/test_importer_loader.py
git commit -m "feat(meeting-scheduler): organized-app JSON backup loader with schema validation (F81.0 task 4)"

Task 5: Mapper `PersonType → PersonRecord`

Files:

Create: packages/jw-meeting-scheduler/src/jw_meeting_scheduler/importer/person_mapper.py
Create: packages/jw-meeting-scheduler/tests/test_person_mapper.py

Interfaces:

Consumes: PersonType (from jw_core), FieldEncryptor, slugify helper.
Produces: map_person(person: PersonType, *, encryptor: FieldEncryptor) -> PersonRecord.
Step 1: Write the failing test

# packages/jw-meeting-scheduler/tests/test_person_mapper.py
"""Tests for the PersonType → PersonRecord mapper."""

from __future__ import annotations

from pathlib import Path

import pytest

from jw_core.models_organized.assignment import AssignmentCode
from jw_meeting_scheduler.crypto import get_encryptor
from jw_meeting_scheduler.importer.loader import load_backup
from jw_meeting_scheduler.importer.person_mapper import (
    PersonMapError,
    map_person,
    slugify_person_id,
)

FIXTURES = Path(__file__).parent / "fixtures"


def test_slugify_basic() -> None:
    assert slugify_person_id("Juan Pérez") == "juan-perez"
    assert slugify_person_id("María José García") == "maria-jose-garcia"
    assert slugify_person_id("O'Connor") == "oconnor"


def test_slugify_rejects_too_short() -> None:
    with pytest.raises(PersonMapError, match="too short"):
        slugify_person_id("Jo")


def test_map_person_minimal_ok() -> None:
    backup = load_backup(FIXTURES / "backup_minimal.json")
    person = backup.persons[0]
    enc = get_encryptor(passphrase=None, congregation_id="test")
    record = map_person(person, encryptor=enc)
    assert record.person_id == "juan-perez"
    assert record.display_name_ciphered == "Juan Pérez"   # no-op encryption
    assert record.gender == "male"
    assert record.is_midweek_student is True
    assert "ms" in record.privileges
    assert AssignmentCode.MM_BibleReading in record.eligible_assignments
    assert AssignmentCode.MM_StartingConversation in record.eligible_assignments
    assert record.status == "active"
    assert record.imported_from == "organized_app"
    assert record.last_updated == "2026-01-01T00:00:00"


def test_map_person_dedups_eligible_assignments() -> None:
    backup = load_backup(FIXTURES / "backup_minimal.json")
    person = backup.persons[0]
    # Add duplicate code in another assignment entry
    person.person_data.assignments.append(
        type(person.person_data.assignments[0])(
            type="duplicate-entry", updatedAt="2026-01-01T00:00:00", values=[100, 100]
        )
    )
    enc = get_encryptor(passphrase=None, congregation_id="test")
    record = map_person(person, encryptor=enc)
    assert record.eligible_assignments.count(AssignmentCode.MM_BibleReading) == 1


def test_map_person_unknown_gender_when_both_false() -> None:
    backup = load_backup(FIXTURES / "backup_minimal.json")
    person = backup.persons[0]
    person.person_data.male.value = False
    person.person_data.female.value = False
    enc = get_encryptor(passphrase=None, congregation_id="test")
    record = map_person(person, encryptor=enc)
    assert record.gender == "unknown"

Step 2: Run tests (expect FAIL)

Run: .venv/bin/python -m pytest packages/jw-meeting-scheduler/tests/test_person_mapper.py -v

Expected: import error.

Step 3: Implement mapper

# packages/jw-meeting-scheduler/src/jw_meeting_scheduler/importer/person_mapper.py
"""Map organized-app PersonType into the scheduler-flat PersonRecord."""

from __future__ import annotations

import re
import unicodedata
from datetime import date

from jw_core.models_organized.assignment import AssignmentCode
from jw_core.models_organized.person import PersonType
from jw_core.privacy.encryption import FieldEncryptor

from jw_meeting_scheduler.models import (
    Gender,
    PersonRecord,
    Privilege,
    Status,
    TimeAway,
)


class PersonMapError(ValueError):
    """Raised when a PersonType cannot be mapped (slug invalid, etc.)."""


_SLUG_INVALID = re.compile(r"[^a-z0-9]+")


def slugify_person_id(display_name: str) -> str:
    """Convert 'Juan Pérez' → 'juan-perez' for the scheduler person_id key.

    Strips accents (NFD normalize), lowercases, replaces runs of non
    alphanumerics with a single hyphen, strips trailing hyphens.
    """
    normalized = unicodedata.normalize("NFD", display_name)
    ascii_only = "".join(c for c in normalized if unicodedata.category(c) != "Mn")
    lowered = ascii_only.lower()
    slug = _SLUG_INVALID.sub("-", lowered).strip("-")
    if len(slug) < 3:
        raise PersonMapError(f"derived slug {slug!r} is too short (need >=3 chars)")
    return slug


def _derive_gender(person: PersonType) -> Gender:
    male = person.person_data.male.value
    female = person.person_data.female.value
    if male and not female:
        return "male"
    if female and not male:
        return "female"
    return "unknown"


def _derive_status(person: PersonType) -> Status:
    pd = person.person_data
    if pd.publisher_baptized.active.value or pd.publisher_unbaptized.active.value:
        return "active"
    if pd.disqualified.value:
        return "disfellowshipped"
    if pd.archived.value:
        return "inactive"
    return "irregular"


def _active_privileges(person: PersonType) -> list[Privilege]:
    today = date.today().isoformat()
    out: list[Privilege] = []
    for entry in person.person_data.privileges:
        if entry.deleted:
            continue
        # Active if end_date is empty or in the future.
        if not entry.end_date or entry.end_date > today:
            out.append(entry.privilege)
    return sorted(set(out))


def _eligible_assignments(person: PersonType) -> list[AssignmentCode]:
    seen: set[AssignmentCode] = set()
    for entry in person.person_data.assignments:
        for code in entry.values:
            seen.add(code)
    return sorted(seen, key=lambda c: c.value)


def _time_away(person: PersonType) -> list[TimeAway]:
    return [
        TimeAway(
            start=date.fromisoformat(t.start_date),
            end=date.fromisoformat(t.end_date),
            reason=t.comments,
        )
        for t in person.person_data.timeAway
        if not t.deleted
    ]


def _crdt_seed(person: PersonType) -> str:
    """Pick the latest updatedAt seen across PersonData fields as CRDT seed."""
    candidates: list[str] = []
    pd = person.person_data
    candidates.extend(
        [
            pd.person_display_name.updatedAt,
            pd.male.updatedAt,
            pd.female.updatedAt,
            pd.archived.updatedAt,
            pd.disqualified.updatedAt,
            pd.publisher_baptized.active.updatedAt,
            pd.publisher_unbaptized.active.updatedAt,
            pd.midweek_meeting_student.active.updatedAt,
        ]
    )
    for a in pd.assignments:
        candidates.append(a.updatedAt)
    for t in pd.timeAway:
        candidates.append(t.updatedAt)
    return max(candidates) if candidates else "1970-01-01T00:00:00"


def map_person(person: PersonType, *, encryptor: FieldEncryptor) -> PersonRecord:
    display = person.person_data.person_display_name.value
    return PersonRecord(
        person_id=slugify_person_id(display),
        display_name_ciphered=encryptor.encrypt(display),
        gender=_derive_gender(person),
        status=_derive_status(person),
        is_midweek_student=person.person_data.midweek_meeting_student.active.value,
        privileges=_active_privileges(person),
        eligible_assignments=_eligible_assignments(person),
        languages=[],   # populated in Task 7 from congregation.languages fallback
        time_away=_time_away(person),
        last_updated=_crdt_seed(person),
        imported_from="organized_app",
    )

Step 4: Run tests

Run: .venv/bin/python -m pytest packages/jw-meeting-scheduler/tests/test_person_mapper.py -v

Expected: 5 passed.

Step 5: Commit

git add packages/jw-meeting-scheduler/src/jw_meeting_scheduler/importer/person_mapper.py packages/jw-meeting-scheduler/tests/test_person_mapper.py
git commit -m "feat(meeting-scheduler): PersonType → PersonRecord mapper (F81.0 task 5)"

Task 6: Mapper `SchedWeek → AssignmentHistoryEntry[]`

Files:

Create: packages/jw-meeting-scheduler/src/jw_meeting_scheduler/importer/schedule_mapper.py
Create: packages/jw-meeting-scheduler/tests/fixtures/backup_with_schedule.json
Create: packages/jw-meeting-scheduler/tests/test_schedule_mapper.py

Interfaces:

Consumes: backup schedules: list[dict] (typed loosely; organized-app schedules vary).
Produces: map_schedule_week(week_dict: dict, *, person_slugs: dict[str, str]) -> list[AssignmentHistoryEntry].
Step 1: Create the fixture with a populated schedule

{
  "schema_version": "3.0.0",
  "exported_at": "2026-06-15T10:00:00",
  "congregation": {
    "id": "kingdom-hall-test",
    "name": "Kingdom Hall Test",
    "languages": ["es"]
  },
  "persons": [],
  "schedules": [
    {
      "weekOf": "2026-06-08",
      "midweek_meeting": {
        "tgw_bible_reading": {
          "main_hall": {
            "value": "Juan Pérez",
            "updatedAt": "2026-06-01T10:00:00",
            "type": "MM_TGWBibleReading_A"
          }
        },
        "ayf_part1": {
          "main_hall": {
            "student": [
              {"value": "Carlos Ruiz", "updatedAt": "2026-06-01T10:00:00", "type": "MM_AYFPart1_Student_A"}
            ],
            "assistant": [
              {"value": "Pedro Gómez", "updatedAt": "2026-06-01T10:00:00", "type": "MM_AYFPart1_Assistant_A"}
            ]
          },
          "aux_class_1": {
            "student": {"value": "Luis Martín", "updatedAt": "2026-06-01T10:00:00", "type": "MM_AYFPart1_Student_B"},
            "assistant": {"value": "Andrés Soto", "updatedAt": "2026-06-01T10:00:00", "type": "MM_AYFPart1_Assistant_B"}
          }
        },
        "chairman": {"main_hall": {"value": "Anciano Rivera", "updatedAt": "2026-06-01T10:00:00", "type": "MM_Chairman_A"}}
      },
      "weekend_meeting": {
        "speaker": {
          "part_1": [{"value": "Hermano González", "updatedAt": "2026-06-01T10:00:00", "type": "WM_Speaker_Part1"}],
          "part_2": [],
          "substitute": []
        },
        "wt_study": {
          "conductor": [{"value": "Anciano Salas", "updatedAt": "2026-06-01T10:00:00", "type": "WM_WTStudy_Conductor"}],
          "reader": [{"value": "Pedro Gómez", "updatedAt": "2026-06-01T10:00:00", "type": "WM_WTStudy_Reader"}]
        }
      }
    }
  ]
}

Step 2: Write the failing test

# packages/jw-meeting-scheduler/tests/test_schedule_mapper.py
"""Tests for the SchedWeek → AssignmentHistoryEntry[] mapper."""

from __future__ import annotations

from datetime import date
from pathlib import Path

from jw_meeting_scheduler.importer.loader import load_backup
from jw_meeting_scheduler.importer.schedule_mapper import map_schedule_week

FIXTURES = Path(__file__).parent / "fixtures"


def _slug_table() -> dict[str, str]:
    return {
        "Juan Pérez": "juan-perez",
        "Carlos Ruiz": "carlos-ruiz",
        "Pedro Gómez": "pedro-gomez",
        "Luis Martín": "luis-martin",
        "Andrés Soto": "andres-soto",
        "Anciano Rivera": "anciano-rivera",
        "Hermano González": "hermano-gonzalez",
        "Anciano Salas": "anciano-salas",
    }


def test_map_schedule_week_extracts_all_populated_slots() -> None:
    backup = load_backup(FIXTURES / "backup_with_schedule.json")
    week = backup.schedules[0]
    entries = map_schedule_week(week, person_slugs=_slug_table())
    fields = {e.assignment_field for e in entries}
    # Reading + AYF + chairman + speaker + WT conductor + WT reader
    assert "MM_TGWBibleReading_A" in fields
    assert "MM_AYFPart1_Student_A" in fields
    assert "MM_AYFPart1_Assistant_A" in fields
    assert "MM_AYFPart1_Student_B" in fields
    assert "MM_AYFPart1_Assistant_B" in fields
    assert "MM_Chairman_A" in fields
    assert "WM_Speaker_Part1" in fields
    assert "WM_WTStudy_Conductor" in fields
    assert "WM_WTStudy_Reader" in fields


def test_map_schedule_week_aux_class_label() -> None:
    backup = load_backup(FIXTURES / "backup_with_schedule.json")
    week = backup.schedules[0]
    entries = map_schedule_week(week, person_slugs=_slug_table())
    aux = [e for e in entries if e.assignment_field.endswith("_B")]
    assert all(e.aula == "aux_class_1" for e in aux)


def test_map_schedule_week_skips_unknown_person() -> None:
    backup = load_backup(FIXTURES / "backup_with_schedule.json")
    week = backup.schedules[0]
    # Strip "Carlos Ruiz" from slug table — that slot is silently skipped.
    table = _slug_table()
    del table["Carlos Ruiz"]
    entries = map_schedule_week(week, person_slugs=table)
    student_a = [e for e in entries if e.assignment_field == "MM_AYFPart1_Student_A"]
    assert student_a == []


def test_map_schedule_week_uses_weekOf_as_meeting_date() -> None:
    backup = load_backup(FIXTURES / "backup_with_schedule.json")
    entries = map_schedule_week(backup.schedules[0], person_slugs=_slug_table())
    assert all(e.meeting_date == date(2026, 6, 8) for e in entries)

Step 3: Run tests (expect FAIL)

Run: .venv/bin/python -m pytest packages/jw-meeting-scheduler/tests/test_schedule_mapper.py -v

Expected: import error.

Step 4: Implement schedule mapper

# packages/jw-meeting-scheduler/src/jw_meeting_scheduler/importer/schedule_mapper.py
"""Map an organized-app week dict into AssignmentHistoryEntry[].

The week structure in organized-app backups is nested and partially
optional. We walk it pragmatically — anything we recognise becomes an
entry; anything we don't is silently skipped (warnings emitted by caller).
"""

from __future__ import annotations

import uuid
from datetime import date
from typing import Any

from jw_core.models_organized.assignment import AssignmentCode

from jw_meeting_scheduler.models import AssignmentHistoryEntry, Aula


# Map assignment_field → AssignmentCode category.
_FIELD_TO_CODE: dict[str, AssignmentCode] = {
    "MM_Chairman_A": AssignmentCode.MM_Chairman,
    "MM_Chairman_B": AssignmentCode.MM_Chairman,
    "MM_OpeningPrayer": AssignmentCode.MM_Prayer,
    "MM_TGWTalk": AssignmentCode.MM_TGWTalk,
    "MM_TGWGems": AssignmentCode.MM_TGWGems,
    "MM_TGWBibleReading_A": AssignmentCode.MM_BibleReading,
    "MM_TGWBibleReading_B": AssignmentCode.MM_BibleReading,
    **{f"MM_AYFPart{n}_Student_{ab}": AssignmentCode.MM_InitialCall
       for n in (1, 2, 3, 4) for ab in ("A", "B")},
    **{f"MM_AYFPart{n}_Assistant_{ab}": AssignmentCode.MM_AssistantOnly
       for n in (1, 2, 3, 4) for ab in ("A", "B")},
    "MM_LCPart1": AssignmentCode.MM_LCPart,
    "MM_LCPart2": AssignmentCode.MM_LCPart,
    "MM_LCPart3": AssignmentCode.MM_LCPart,
    "MM_LCCBSConductor": AssignmentCode.MM_CBSConductor,
    "MM_LCCBSReader": AssignmentCode.MM_CBSReader,
    "MM_ClosingPrayer": AssignmentCode.MM_Prayer,
    "WM_Chairman": AssignmentCode.WM_Chairman,
    "WM_OpeningPrayer": AssignmentCode.WM_Prayer,
    "WM_Speaker_Part1": AssignmentCode.WM_Speaker,
    "WM_Speaker_Part2": AssignmentCode.WM_Speaker,
    "WM_WTStudy_Conductor": AssignmentCode.WM_WTStudyConductor,
    "WM_WTStudy_Reader": AssignmentCode.WM_WTStudyReader,
    "WM_ClosingPrayer": AssignmentCode.WM_Prayer,
    "WM_SubstituteSpeaker": AssignmentCode.WM_Speaker,
}


def _aula_for_field(field: str) -> Aula:
    if field.endswith("_B"):
        return "aux_class_1"
    if field.endswith("_C"):
        return "aux_class_2"
    return "main_hall"


def _slot_to_entry(
    slot: dict[str, Any],
    *,
    week_of: date,
    person_slugs: dict[str, str],
) -> AssignmentHistoryEntry | None:
    name = slot.get("value", "")
    field = slot.get("type", "")
    updated_at = slot.get("updatedAt", "")
    if not name or not field or field not in _FIELD_TO_CODE:
        return None
    person_id = person_slugs.get(name)
    if not person_id:
        return None
    return AssignmentHistoryEntry(
        entry_id=str(uuid.uuid5(uuid.NAMESPACE_DNS, f"{field}|{name}|{week_of.isoformat()}")),
        person_id=person_id,
        assignment_field=field,
        assignment_code=_FIELD_TO_CODE[field],
        meeting_date=week_of,
        aula=_aula_for_field(field),
        confirmed=True,                    # historical = was on the actual schedule
        confirmed_at=updated_at,
        updated_at=updated_at,
    )


def _walk(obj: Any, *, week_of: date, person_slugs: dict[str, str]) -> list[AssignmentHistoryEntry]:
    """Depth-first walk of the schedule tree picking up dict slots."""
    out: list[AssignmentHistoryEntry] = []
    if isinstance(obj, dict):
        if "value" in obj and "type" in obj and "updatedAt" in obj:
            entry = _slot_to_entry(obj, week_of=week_of, person_slugs=person_slugs)
            if entry:
                out.append(entry)
            return out
        for v in obj.values():
            out.extend(_walk(v, week_of=week_of, person_slugs=person_slugs))
    elif isinstance(obj, list):
        for v in obj:
            out.extend(_walk(v, week_of=week_of, person_slugs=person_slugs))
    return out


def map_schedule_week(
    week_dict: dict[str, Any],
    *,
    person_slugs: dict[str, str],
) -> list[AssignmentHistoryEntry]:
    """Flatten one organized-app weekly schedule into AssignmentHistoryEntry[]."""
    week_of_raw = week_dict.get("weekOf")
    if not week_of_raw:
        return []
    week_of = date.fromisoformat(week_of_raw)
    return _walk(week_dict, week_of=week_of, person_slugs=person_slugs)

Step 5: Run tests

Run: .venv/bin/python -m pytest packages/jw-meeting-scheduler/tests/test_schedule_mapper.py -v

Expected: 4 passed.

Step 6: Commit

git add packages/jw-meeting-scheduler/src/jw_meeting_scheduler/importer/schedule_mapper.py packages/jw-meeting-scheduler/tests/fixtures/backup_with_schedule.json packages/jw-meeting-scheduler/tests/test_schedule_mapper.py
git commit -m "feat(meeting-scheduler): SchedWeek → AssignmentHistoryEntry[] flattening mapper (F81.0 task 6)"

Task 7: Store SQLite + persistencia + idempotencia

Files:

Create: packages/jw-meeting-scheduler/src/jw_meeting_scheduler/store/__init__.py
Create: packages/jw-meeting-scheduler/src/jw_meeting_scheduler/store/db.py
Create: packages/jw-meeting-scheduler/tests/test_store.py

Interfaces:

Consumes: PersonRecord, AssignmentHistoryEntry.
Produces: open_store(congregation_id, *, root=None) -> SchedulerStore with upsert_person, get_person, list_people, record_history, last_history_for, slug_table().
Step 1: Write the failing test

# packages/jw-meeting-scheduler/tests/test_store.py
"""SchedulerStore SQLite roundtrip + idempotency + CRDT respect."""

from __future__ import annotations

from datetime import date
from pathlib import Path

import pytest

from jw_core.models_organized.assignment import AssignmentCode
from jw_meeting_scheduler.models import AssignmentHistoryEntry, PersonRecord
from jw_meeting_scheduler.store import open_store


def _record(slug: str = "juan-perez", updated: str = "2026-06-17T10:00:00") -> PersonRecord:
    return PersonRecord(
        person_id=slug,
        display_name_ciphered="Juan Pérez",
        gender="male",
        last_updated=updated,
        imported_from="organized_app",
    )


def test_open_store_creates_db(tmp_path: Path) -> None:
    store = open_store("kingdom-hall-test", root=tmp_path)
    assert (tmp_path / "kingdom-hall-test").is_dir()
    assert store.path.exists()


def test_upsert_and_get_person(tmp_path: Path) -> None:
    store = open_store("cong-a", root=tmp_path)
    store.upsert_person(_record())
    fetched = store.get_person("juan-perez")
    assert fetched is not None
    assert fetched.display_name_ciphered == "Juan Pérez"


def test_upsert_is_idempotent_when_same_timestamp(tmp_path: Path) -> None:
    store = open_store("cong-b", root=tmp_path)
    rec = _record(updated="2026-06-17T10:00:00")
    store.upsert_person(rec)
    store.upsert_person(rec)
    assert len(store.list_people()) == 1


def test_upsert_respects_crdt_keeps_local_when_newer(tmp_path: Path) -> None:
    store = open_store("cong-c", root=tmp_path)
    newer = _record(updated="2026-06-17T10:00:00")
    older = _record(updated="2026-01-01T00:00:00")
    store.upsert_person(newer)
    # Re-import older shouldn't overwrite.
    store.upsert_person(older)
    fetched = store.get_person("juan-perez")
    assert fetched is not None
    assert fetched.last_updated == "2026-06-17T10:00:00"


def test_record_history_unique_by_entry_id(tmp_path: Path) -> None:
    store = open_store("cong-d", root=tmp_path)
    entry = AssignmentHistoryEntry(
        entry_id="hist-1",
        person_id="juan-perez",
        assignment_field="MM_TGWBibleReading_A",
        assignment_code=AssignmentCode.MM_BibleReading,
        meeting_date=date(2026, 6, 8),
        updated_at="2026-06-01T10:00:00",
    )
    store.record_history(entry)
    store.record_history(entry)   # second time is no-op
    assert len(store.list_history("juan-perez")) == 1


def test_last_history_for_returns_most_recent(tmp_path: Path) -> None:
    store = open_store("cong-e", root=tmp_path)
    e1 = AssignmentHistoryEntry(
        entry_id="hist-1",
        person_id="juan-perez",
        assignment_field="MM_TGWBibleReading_A",
        assignment_code=AssignmentCode.MM_BibleReading,
        meeting_date=date(2026, 4, 1),
        updated_at="2026-04-01T00:00:00",
    )
    e2 = AssignmentHistoryEntry(
        entry_id="hist-2",
        person_id="juan-perez",
        assignment_field="MM_TGWBibleReading_A",
        assignment_code=AssignmentCode.MM_BibleReading,
        meeting_date=date(2026, 6, 1),
        updated_at="2026-06-01T00:00:00",
    )
    store.record_history(e1)
    store.record_history(e2)
    last = store.last_history_for("juan-perez", AssignmentCode.MM_BibleReading)
    assert last is not None
    assert last.meeting_date == date(2026, 6, 1)


def test_slug_table_round_trips(tmp_path: Path) -> None:
    store = open_store("cong-f", root=tmp_path)
    store.upsert_person(_record(slug="juan-perez"))
    store.upsert_person(_record(slug="carlos-ruiz"))
    table = store.slug_table()
    assert table["Juan Pérez"] == "juan-perez"
    assert table["Juan Pérez"] != table.get("Carlos Ruiz")

Step 2: Run tests (expect FAIL)

Run: .venv/bin/python -m pytest packages/jw-meeting-scheduler/tests/test_store.py -v

Expected: import error.

Step 3: Implement store

# packages/jw-meeting-scheduler/src/jw_meeting_scheduler/store/__init__.py
"""SchedulerStore: SQLite-backed people + history."""

from jw_meeting_scheduler.store.db import SchedulerStore, open_store

__all__ = ["SchedulerStore", "open_store"]

# packages/jw-meeting-scheduler/src/jw_meeting_scheduler/store/db.py
"""SQLite store for PersonRecord + AssignmentHistoryEntry.

Encryption is opt-in via FieldEncryptor (see jw_meeting_scheduler.crypto);
fields holding PII are already ciphered by the importer before they hit
the store. The store stays agnostic of the cipher state — strings in,
strings out.
"""

from __future__ import annotations

import json
import os
import sqlite3
from dataclasses import dataclass
from datetime import date
from pathlib import Path

from jw_core.models_organized.assignment import AssignmentCode

from jw_meeting_scheduler.models import AssignmentHistoryEntry, PersonRecord


_SCHEMA = """
CREATE TABLE IF NOT EXISTS persons (
    person_id TEXT PRIMARY KEY,
    display_name_ciphered TEXT NOT NULL,
    gender TEXT NOT NULL,
    status TEXT NOT NULL,
    is_midweek_student INTEGER NOT NULL DEFAULT 0,
    privileges_json TEXT NOT NULL DEFAULT '[]',
    eligible_codes_json TEXT NOT NULL DEFAULT '[]',
    skill_level_json TEXT NOT NULL DEFAULT '{}',
    languages_json TEXT NOT NULL DEFAULT '[]',
    time_away_json TEXT NOT NULL DEFAULT '[]',
    last_updated TEXT NOT NULL,
    imported_from TEXT NOT NULL DEFAULT 'manual'
);

CREATE TABLE IF NOT EXISTS history (
    entry_id TEXT PRIMARY KEY,
    person_id TEXT NOT NULL,
    assignment_field TEXT NOT NULL,
    assignment_code INTEGER NOT NULL,
    meeting_date TEXT NOT NULL,
    aula TEXT NOT NULL DEFAULT 'main_hall',
    confirmed INTEGER NOT NULL DEFAULT 0,
    confirmed_at TEXT,
    cancelled INTEGER NOT NULL DEFAULT 0,
    cancellation_reason TEXT NOT NULL DEFAULT '',
    updated_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_history_person_code_date
    ON history (person_id, assignment_code, meeting_date DESC);
"""


def _default_root() -> Path:
    return Path(os.getenv("JW_MEETING_SCHED_HOME", "~/.jw-agent-toolkit/congregations")).expanduser()


@dataclass(frozen=True)
class SchedulerStore:
    congregation_id: str
    path: Path
    _conn: sqlite3.Connection

    # ----- people ------------------------------------------------------

    def upsert_person(self, rec: PersonRecord) -> None:
        existing = self.get_person(rec.person_id)
        if existing is not None and existing.last_updated >= rec.last_updated:
            return  # CRDT: local newer or same → keep local
        self._conn.execute(
            """
            INSERT INTO persons (
                person_id, display_name_ciphered, gender, status, is_midweek_student,
                privileges_json, eligible_codes_json, skill_level_json,
                languages_json, time_away_json, last_updated, imported_from
            ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)
            ON CONFLICT(person_id) DO UPDATE SET
                display_name_ciphered=excluded.display_name_ciphered,
                gender=excluded.gender,
                status=excluded.status,
                is_midweek_student=excluded.is_midweek_student,
                privileges_json=excluded.privileges_json,
                eligible_codes_json=excluded.eligible_codes_json,
                skill_level_json=excluded.skill_level_json,
                languages_json=excluded.languages_json,
                time_away_json=excluded.time_away_json,
                last_updated=excluded.last_updated,
                imported_from=excluded.imported_from
            """,
            (
                rec.person_id,
                rec.display_name_ciphered,
                rec.gender,
                rec.status,
                1 if rec.is_midweek_student else 0,
                json.dumps(rec.privileges),
                json.dumps([c.value for c in rec.eligible_assignments]),
                json.dumps({str(k.value): v for k, v in rec.skill_level.items()}),
                json.dumps(rec.languages),
                json.dumps([t.model_dump(mode="json") for t in rec.time_away]),
                rec.last_updated,
                rec.imported_from,
            ),
        )
        self._conn.commit()

    def get_person(self, person_id: str) -> PersonRecord | None:
        row = self._conn.execute(
            "SELECT * FROM persons WHERE person_id = ?", (person_id,)
        ).fetchone()
        if row is None:
            return None
        return _row_to_person(row)

    def list_people(self) -> list[PersonRecord]:
        rows = self._conn.execute("SELECT * FROM persons ORDER BY person_id").fetchall()
        return [_row_to_person(r) for r in rows]

    def slug_table(self) -> dict[str, str]:
        """Reverse lookup display_name → person_id for the schedule mapper.

        NOTE: when display_name_ciphered is no-op (no JW_PRIVACY_KEY) the
        ciphertext IS the plaintext, so the table works directly. With a
        real key the caller must pass a decrypted view (out of scope here).
        """
        return {r.display_name_ciphered: r.person_id for r in self.list_people()}

    # ----- history -----------------------------------------------------

    def record_history(self, entry: AssignmentHistoryEntry) -> None:
        self._conn.execute(
            """
            INSERT OR IGNORE INTO history (
                entry_id, person_id, assignment_field, assignment_code,
                meeting_date, aula, confirmed, confirmed_at,
                cancelled, cancellation_reason, updated_at
            ) VALUES (?,?,?,?,?,?,?,?,?,?,?)
            """,
            (
                entry.entry_id,
                entry.person_id,
                entry.assignment_field,
                entry.assignment_code.value,
                entry.meeting_date.isoformat(),
                entry.aula,
                1 if entry.confirmed else 0,
                entry.confirmed_at,
                1 if entry.cancelled else 0,
                entry.cancellation_reason,
                entry.updated_at,
            ),
        )
        self._conn.commit()

    def list_history(self, person_id: str) -> list[AssignmentHistoryEntry]:
        rows = self._conn.execute(
            "SELECT * FROM history WHERE person_id = ? ORDER BY meeting_date DESC",
            (person_id,),
        ).fetchall()
        return [_row_to_history(r) for r in rows]

    def last_history_for(
        self, person_id: str, code: AssignmentCode
    ) -> AssignmentHistoryEntry | None:
        row = self._conn.execute(
            """
            SELECT * FROM history
            WHERE person_id = ? AND assignment_code = ?
            ORDER BY meeting_date DESC
            LIMIT 1
            """,
            (person_id, code.value),
        ).fetchone()
        return _row_to_history(row) if row else None


def open_store(congregation_id: str, *, root: Path | None = None) -> SchedulerStore:
    root = root or _default_root()
    folder = root / congregation_id
    folder.mkdir(parents=True, exist_ok=True)
    db_path = folder / "scheduler.db"
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    conn.executescript(_SCHEMA)
    return SchedulerStore(congregation_id=congregation_id, path=db_path, _conn=conn)


def _row_to_person(row: sqlite3.Row) -> PersonRecord:
    from jw_meeting_scheduler.models import TimeAway

    return PersonRecord(
        person_id=row["person_id"],
        display_name_ciphered=row["display_name_ciphered"],
        gender=row["gender"],
        status=row["status"],
        is_midweek_student=bool(row["is_midweek_student"]),
        privileges=json.loads(row["privileges_json"]),
        eligible_assignments=[AssignmentCode(c) for c in json.loads(row["eligible_codes_json"])],
        skill_level={AssignmentCode(int(k)): v for k, v in json.loads(row["skill_level_json"]).items()},
        languages=json.loads(row["languages_json"]),
        time_away=[TimeAway.model_validate(t) for t in json.loads(row["time_away_json"])],
        last_updated=row["last_updated"],
        imported_from=row["imported_from"],
    )


def _row_to_history(row: sqlite3.Row) -> AssignmentHistoryEntry:
    return AssignmentHistoryEntry(
        entry_id=row["entry_id"],
        person_id=row["person_id"],
        assignment_field=row["assignment_field"],
        assignment_code=AssignmentCode(row["assignment_code"]),
        meeting_date=date.fromisoformat(row["meeting_date"]),
        aula=row["aula"],
        confirmed=bool(row["confirmed"]),
        confirmed_at=row["confirmed_at"],
        cancelled=bool(row["cancelled"]),
        cancellation_reason=row["cancellation_reason"],
        updated_at=row["updated_at"],
    )

Step 4: Run tests

Run: .venv/bin/python -m pytest packages/jw-meeting-scheduler/tests/test_store.py -v

Expected: 7 passed.

Step 5: Commit

git add packages/jw-meeting-scheduler/src/jw_meeting_scheduler/store/ packages/jw-meeting-scheduler/tests/test_store.py
git commit -m "feat(meeting-scheduler): SQLite store with CRDT-aware upsert and history indices (F81.0 task 7)"

Task 8: Dry-run + diff helpers

Files:

Create: packages/jw-meeting-scheduler/src/jw_meeting_scheduler/importer/diff.py
Create: packages/jw-meeting-scheduler/tests/test_diff.py

Interfaces:

Consumes: SchedulerStore, PersonRecord.
Produces: ImportDiff(added: list[str], updated: list[str], kept_local: list[str], unchanged: list[str]) and compute_person_diff(store, incoming) -> ImportDiff.
Step 1: Write the failing test

# packages/jw-meeting-scheduler/tests/test_diff.py
"""Tests for the import diff computation."""

from __future__ import annotations

from pathlib import Path

from jw_meeting_scheduler.importer.diff import compute_person_diff
from jw_meeting_scheduler.models import PersonRecord
from jw_meeting_scheduler.store import open_store


def _rec(slug: str, updated: str = "2026-06-01T00:00:00") -> PersonRecord:
    return PersonRecord(
        person_id=slug,
        display_name_ciphered=slug,
        gender="male",
        last_updated=updated,
    )


def test_diff_empty_store_classifies_all_as_added(tmp_path: Path) -> None:
    store = open_store("cong-x", root=tmp_path)
    diff = compute_person_diff(store, [_rec("juan-perez"), _rec("carlos-ruiz")])
    assert sorted(diff.added) == ["carlos-ruiz", "juan-perez"]
    assert diff.updated == []
    assert diff.kept_local == []


def test_diff_classifies_newer_as_updated(tmp_path: Path) -> None:
    store = open_store("cong-y", root=tmp_path)
    store.upsert_person(_rec("juan-perez", updated="2026-01-01T00:00:00"))
    diff = compute_person_diff(store, [_rec("juan-perez", updated="2026-06-01T00:00:00")])
    assert diff.updated == ["juan-perez"]
    assert diff.kept_local == []


def test_diff_classifies_older_as_kept_local(tmp_path: Path) -> None:
    store = open_store("cong-z", root=tmp_path)
    store.upsert_person(_rec("juan-perez", updated="2026-06-01T00:00:00"))
    diff = compute_person_diff(store, [_rec("juan-perez", updated="2026-01-01T00:00:00")])
    assert diff.kept_local == ["juan-perez"]


def test_diff_same_timestamp_classifies_unchanged(tmp_path: Path) -> None:
    store = open_store("cong-w", root=tmp_path)
    store.upsert_person(_rec("juan-perez", updated="2026-06-01T00:00:00"))
    diff = compute_person_diff(store, [_rec("juan-perez", updated="2026-06-01T00:00:00")])
    assert diff.unchanged == ["juan-perez"]

Step 2: Run tests (expect FAIL)

Run: .venv/bin/python -m pytest packages/jw-meeting-scheduler/tests/test_diff.py -v

Expected: import error.

Step 3: Implement diff

# packages/jw-meeting-scheduler/src/jw_meeting_scheduler/importer/diff.py
"""Pre-write diff for the import pipeline: what would change vs the store."""

from __future__ import annotations

from dataclasses import dataclass, field

from jw_meeting_scheduler.models import PersonRecord
from jw_meeting_scheduler.store import SchedulerStore


@dataclass(frozen=True)
class ImportDiff:
    added: list[str] = field(default_factory=list)
    updated: list[str] = field(default_factory=list)
    kept_local: list[str] = field(default_factory=list)
    unchanged: list[str] = field(default_factory=list)


def compute_person_diff(
    store: SchedulerStore, incoming: list[PersonRecord]
) -> ImportDiff:
    added: list[str] = []
    updated: list[str] = []
    kept_local: list[str] = []
    unchanged: list[str] = []
    for rec in incoming:
        existing = store.get_person(rec.person_id)
        if existing is None:
            added.append(rec.person_id)
        elif rec.last_updated > existing.last_updated:
            updated.append(rec.person_id)
        elif rec.last_updated < existing.last_updated:
            kept_local.append(rec.person_id)
        else:
            unchanged.append(rec.person_id)
    return ImportDiff(
        added=sorted(added),
        updated=sorted(updated),
        kept_local=sorted(kept_local),
        unchanged=sorted(unchanged),
    )

Step 4: Run tests

Run: .venv/bin/python -m pytest packages/jw-meeting-scheduler/tests/test_diff.py -v

Expected: 4 passed.

Step 5: Commit

git add packages/jw-meeting-scheduler/src/jw_meeting_scheduler/importer/diff.py packages/jw-meeting-scheduler/tests/test_diff.py
git commit -m "feat(meeting-scheduler): pre-write import diff classifies added/updated/kept-local (F81.0 task 8)"

Task 9: Pipeline E2E `run_import()` + dry-run flag

Files:

Create: packages/jw-meeting-scheduler/src/jw_meeting_scheduler/importer/pipeline.py
Create: packages/jw-meeting-scheduler/tests/test_pipeline.py

Interfaces:

Consumes: loader + person_mapper + schedule_mapper + diff + store + crypto.
Produces: ImportReport(persons: ImportDiff, history_added: int, history_skipped: int) and run_import(backup_path, *, congregation_id, store, encryptor, dry_run) -> ImportReport.
Step 1: Write the failing test

# packages/jw-meeting-scheduler/tests/test_pipeline.py
"""Integration test: full import pipeline end-to-end."""

from __future__ import annotations

from pathlib import Path

from jw_meeting_scheduler.crypto import get_encryptor
from jw_meeting_scheduler.importer.pipeline import run_import
from jw_meeting_scheduler.store import open_store

FIXTURES = Path(__file__).parent / "fixtures"


def test_dry_run_does_not_write(tmp_path: Path) -> None:
    store = open_store("cong-dry", root=tmp_path)
    enc = get_encryptor(passphrase=None, congregation_id="cong-dry")
    report = run_import(
        FIXTURES / "backup_minimal.json",
        congregation_id="cong-dry",
        store=store,
        encryptor=enc,
        dry_run=True,
    )
    assert report.persons.added == ["juan-perez"]
    assert store.list_people() == []


def test_real_import_persists(tmp_path: Path) -> None:
    store = open_store("cong-real", root=tmp_path)
    enc = get_encryptor(passphrase=None, congregation_id="cong-real")
    run_import(
        FIXTURES / "backup_minimal.json",
        congregation_id="cong-real",
        store=store,
        encryptor=enc,
        dry_run=False,
    )
    assert len(store.list_people()) == 1
    rec = store.get_person("juan-perez")
    assert rec is not None
    assert rec.imported_from == "organized_app"


def test_re_import_is_idempotent(tmp_path: Path) -> None:
    store = open_store("cong-idem", root=tmp_path)
    enc = get_encryptor(passphrase=None, congregation_id="cong-idem")
    run_import(FIXTURES / "backup_minimal.json", congregation_id="cong-idem", store=store, encryptor=enc, dry_run=False)
    run_import(FIXTURES / "backup_minimal.json", congregation_id="cong-idem", store=store, encryptor=enc, dry_run=False)
    assert len(store.list_people()) == 1


def test_history_imports_from_schedule(tmp_path: Path) -> None:
    store = open_store("cong-hist", root=tmp_path)
    enc = get_encryptor(passphrase=None, congregation_id="cong-hist")
    # Pre-seed the people referenced in the schedule (otherwise mapper skips them).
    from jw_meeting_scheduler.models import PersonRecord

    for slug, display in [
        ("juan-perez", "Juan Pérez"),
        ("carlos-ruiz", "Carlos Ruiz"),
        ("pedro-gomez", "Pedro Gómez"),
        ("luis-martin", "Luis Martín"),
        ("andres-soto", "Andrés Soto"),
        ("anciano-rivera", "Anciano Rivera"),
        ("hermano-gonzalez", "Hermano González"),
        ("anciano-salas", "Anciano Salas"),
    ]:
        store.upsert_person(
            PersonRecord(
                person_id=slug,
                display_name_ciphered=display,
                gender="male",
                last_updated="2026-01-01T00:00:00",
            )
        )
    report = run_import(
        FIXTURES / "backup_with_schedule.json",
        congregation_id="cong-hist",
        store=store,
        encryptor=enc,
        dry_run=False,
    )
    assert report.history_added > 0
    juan = store.list_history("juan-perez")
    assert any(e.assignment_field == "MM_TGWBibleReading_A" for e in juan)

Step 2: Run tests (expect FAIL)

Run: .venv/bin/python -m pytest packages/jw-meeting-scheduler/tests/test_pipeline.py -v

Expected: import error.

Step 3: Implement pipeline

# packages/jw-meeting-scheduler/src/jw_meeting_scheduler/importer/pipeline.py
"""End-to-end orchestration of the organized-app import.

  1. Load backup JSON.
  2. Map persons → PersonRecord[].
  3. Compute diff vs current store.
  4. If not dry_run, upsert persons; CRDT-protected.
  5. Build slug_table from final store (includes pre-existing manual people).
  6. Map each weekly schedule → AssignmentHistoryEntry[]; record idempotently.
"""

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path

from jw_core.privacy.encryption import FieldEncryptor

from jw_meeting_scheduler.importer.diff import ImportDiff, compute_person_diff
from jw_meeting_scheduler.importer.loader import load_backup
from jw_meeting_scheduler.importer.person_mapper import map_person
from jw_meeting_scheduler.importer.schedule_mapper import map_schedule_week
from jw_meeting_scheduler.store import SchedulerStore


@dataclass(frozen=True)
class ImportReport:
    congregation_id: str
    persons: ImportDiff
    history_added: int
    history_skipped: int


def run_import(
    backup_path: Path,
    *,
    congregation_id: str,
    store: SchedulerStore,
    encryptor: FieldEncryptor,
    dry_run: bool = False,
) -> ImportReport:
    backup = load_backup(backup_path)
    records = [map_person(p, encryptor=encryptor) for p in backup.persons]
    diff = compute_person_diff(store, records)

    history_added = 0
    history_skipped = 0
    if not dry_run:
        for rec in records:
            store.upsert_person(rec)
        slug_table = store.slug_table()
        for week_dict in backup.schedules:
            entries = map_schedule_week(week_dict, person_slugs=slug_table)
            for entry in entries:
                before = len(store.list_history(entry.person_id))
                store.record_history(entry)
                after = len(store.list_history(entry.person_id))
                if after > before:
                    history_added += 1
                else:
                    history_skipped += 1

    return ImportReport(
        congregation_id=congregation_id,
        persons=diff,
        history_added=history_added,
        history_skipped=history_skipped,
    )

Step 4: Run tests

Run: .venv/bin/python -m pytest packages/jw-meeting-scheduler/tests/test_pipeline.py -v

Expected: 4 passed.

Step 5: Commit

git add packages/jw-meeting-scheduler/src/jw_meeting_scheduler/importer/pipeline.py packages/jw-meeting-scheduler/tests/test_pipeline.py
git commit -m "feat(meeting-scheduler): import pipeline E2E with dry-run + history idempotency (F81.0 task 9)"

Task 10: CLI command `jw scheduler import`

Files:

Create: packages/jw-cli/src/jw_cli/commands/scheduler.py
Modify: packages/jw-cli/src/jw_cli/main.py (registrar subapp)
Create: packages/jw-cli/tests/test_scheduler_command.py

Interfaces:

Consumes: run_import, open_store, get_encryptor.
Produces: jw scheduler import --backup PATH --congregation ID [--dry-run] [--passphrase TEXT] CLI command.
Step 1: Write the failing test

# packages/jw-cli/tests/test_scheduler_command.py
"""Smoke test for `jw scheduler import` command."""

from __future__ import annotations

from pathlib import Path

from typer.testing import CliRunner

from jw_cli.main import app


FIXTURES = (
    Path(__file__).parent.parent.parent
    / "jw-meeting-scheduler"
    / "tests"
    / "fixtures"
)


def test_jw_scheduler_import_dry_run(tmp_path: Path, monkeypatch) -> None:
    monkeypatch.setenv("JW_MEETING_SCHED_HOME", str(tmp_path))
    runner = CliRunner()
    result = runner.invoke(
        app,
        [
            "scheduler",
            "import",
            "--backup",
            str(FIXTURES / "backup_minimal.json"),
            "--congregation",
            "cli-test",
            "--dry-run",
        ],
    )
    assert result.exit_code == 0, result.stdout
    assert "added" in result.stdout.lower()
    assert "juan-perez" in result.stdout
    # Store dir was created but db should be empty for persons.
    assert (tmp_path / "cli-test" / "scheduler.db").exists()


def test_jw_scheduler_import_real_persists(tmp_path: Path, monkeypatch) -> None:
    monkeypatch.setenv("JW_MEETING_SCHED_HOME", str(tmp_path))
    runner = CliRunner()
    result = runner.invoke(
        app,
        [
            "scheduler",
            "import",
            "--backup",
            str(FIXTURES / "backup_minimal.json"),
            "--congregation",
            "cli-real",
        ],
    )
    assert result.exit_code == 0
    from jw_meeting_scheduler.store import open_store

    store = open_store("cli-real", root=tmp_path)
    assert len(store.list_people()) == 1

Step 2: Run tests (expect FAIL)

Run: .venv/bin/python -m pytest packages/jw-cli/tests/test_scheduler_command.py -v

Expected: command not found / module missing.

Step 3: Implement CLI command

# packages/jw-cli/src/jw_cli/commands/scheduler.py
"""`jw scheduler` subapp — import organized-app backups, etc."""

from __future__ import annotations

from pathlib import Path

import typer
from rich.console import Console
from rich.table import Table

app = typer.Typer(help="Meeting-scheduler operations (import, suggest, confirm).")
console = Console()


@app.command("import")
def import_cmd(
    backup: Path = typer.Option(..., "--backup", exists=True, readable=True),
    congregation: str = typer.Option(..., "--congregation", help="Congregation id (slug)"),
    dry_run: bool = typer.Option(False, "--dry-run", help="Show diff but do not write."),
    passphrase: str | None = typer.Option(
        None, "--passphrase", help="Encryption passphrase. If unset, uses JW_PRIVACY_KEY or no-op."
    ),
) -> None:
    """Import an `organized-app` backup into the local scheduler store."""
    from jw_meeting_scheduler.crypto import get_encryptor
    from jw_meeting_scheduler.importer.pipeline import run_import
    from jw_meeting_scheduler.store import open_store

    store = open_store(congregation)
    encryptor = get_encryptor(passphrase=passphrase, congregation_id=congregation)
    report = run_import(
        backup, congregation_id=congregation, store=store, encryptor=encryptor, dry_run=dry_run
    )
    _render_report(report, dry_run=dry_run)


def _render_report(report, *, dry_run: bool) -> None:
    title = "Import report (dry-run)" if dry_run else "Import report"
    table = Table(title=title)
    table.add_column("Category")
    table.add_column("Count")
    table.add_column("Person ids")
    p = report.persons
    table.add_row("added", str(len(p.added)), ", ".join(p.added))
    table.add_row("updated", str(len(p.updated)), ", ".join(p.updated))
    table.add_row("kept_local", str(len(p.kept_local)), ", ".join(p.kept_local))
    table.add_row("unchanged", str(len(p.unchanged)), ", ".join(p.unchanged))
    table.add_row("history_added", str(report.history_added), "")
    table.add_row("history_skipped", str(report.history_skipped), "")
    console.print(table)

Step 4: Register subapp in main.py

Edit packages/jw-cli/src/jw_cli/main.py to add:

from jw_cli.commands import scheduler as scheduler_cmd

app.add_typer(scheduler_cmd.app, name="scheduler")

(Insertar junto a los otros app.add_typer(...) que ya existen, manteniendo orden alfabético si lo hay.)

Step 5: Add jw-meeting-scheduler as dep of jw-cli

Edit packages/jw-cli/pyproject.toml, sección dependencies:

dependencies = [
    # ... existentes ...
    "jw-meeting-scheduler",
]

Run: uv sync --all-packages.

Step 6: Run tests

Run: .venv/bin/python -m pytest packages/jw-cli/tests/test_scheduler_command.py -v

Expected: 2 passed.

Step 7: Commit

git add packages/jw-cli/src/jw_cli/commands/scheduler.py packages/jw-cli/src/jw_cli/main.py packages/jw-cli/pyproject.toml packages/jw-cli/tests/test_scheduler_command.py uv.lock
git commit -m "feat(cli): jw scheduler import command with dry-run + Rich table report (F81.0 task 10)"

Task 11: Documentación operativa

Files:

Create: docs/guias/meeting-scheduler-import.md
Modify: docs/ROADMAP.md (marcar F81.0 ✅)

Interfaces:

Consumes: nothing.
Produces: guía paso a paso del flujo de import.
Step 1: Write the guide

<!-- docs/guias/meeting-scheduler-import.md -->
# Importar un backup de organized-app

Esta guía cubre F81.0: cómo poblar el store del scheduler a partir
de un backup JSON exportado desde la web app `organized-app`.

## Requisitos

- `uv sync --all-packages` corrido al menos una vez.
- Backup JSON exportado de organized-app (Settings → Backup → Export).
- (opcional) `JW_PRIVACY_KEY` exportada o `--passphrase` listo.

## Comando

```bash
# Dry-run: muestra qué cambiaría sin tocar el store
uv run jw scheduler import \
  --backup ~/Downloads/organized-backup.json \
  --congregation kingdom-hall-central \
  --dry-run

# Import real
uv run jw scheduler import \
  --backup ~/Downloads/organized-backup.json \
  --congregation kingdom-hall-central \
  --passphrase "correct-horse-battery-staple"

Qué pasa por dentro

Lee el JSON con jw_meeting_scheduler.importer.loader.load_backup.
Por cada PersonType corre map_person → PersonRecord.
Calcula diff vs el store (compute_person_diff):
- added: el slug no existía.
- updated: el slug existía con last_updated anterior.
- kept_local: el slug existía con last_updated posterior → no se sobrescribe (CRDT respect).
- unchanged: timestamps iguales.
Si no es dry-run, upserta personas y luego por cada SchedWeek ejecuta map_schedule_week → AssignmentHistoryEntry[] y los inserta con INSERT OR IGNORE (idempotente por entry_id).

Ubicación del store

~/.jw-agent-toolkit/congregations/<congregation_id>/scheduler.db.

Override con env var JW_MEETING_SCHED_HOME o passa --root (futuro).

Cifrado

display_name_ciphered se cifra con jw_core.privacy.encryption.FieldEncryptor. Llave en orden:

--passphrase → derivada vía PBKDF2-HMAC-SHA256 (200k iters) con salt "jw-meeting-scheduler/v1:<congregation_id>".
JW_PRIVACY_KEY (urlsafe base64 32 bytes).
Sin llave → no-op + warning (cleartext en disco).

Re-import

Repetir el comando es seguro. CRDT por last_updated y INSERT OR IGNORE por entry_id garantizan que no se duplica ni se machaca ediciones manuales.

Próximos pasos (F81.1+)

Edición manual de personas con jw scheduler person edit ... (F81.1).
YAML de restricciones con jw scheduler constraints init (F81.2).
Solver CP-SAT con jw scheduler suggest --week ... (F81.3+).


- [ ] **Step 2: Marcar F81.0 ✅ en ROADMAP**

Edit la línea correspondiente en `docs/ROADMAP.md`:

```markdown
- ⬜ **F81.0 — importador `organized-app`** (1 semana): JSON backup →

cambiar ⬜ por ✅ y añadir (entregado YYYY-MM-DD).

Step 3: Smoke-test del repo completo

Run: .venv/bin/python -m pytest packages/jw-meeting-scheduler/tests/ packages/jw-cli/tests/test_scheduler_command.py -v

Expected: TODOS los tests verdes.

Step 4: Smoke-test CLI E2E real

Run: uv run jw scheduler import --backup packages/jw-meeting-scheduler/tests/fixtures/backup_minimal.json --congregation smoke-test --dry-run

Expected: tabla Rich impresa, exit code 0, sin tocar el filesystem fuera de ~/.jw-agent-toolkit/.

Step 5: Commit

git add docs/guias/meeting-scheduler-import.md docs/ROADMAP.md
git commit -m "docs(meeting-scheduler): import flow guide + mark F81.0 delivered (F81.0 task 11)"

Self-review (al cerrar el plan)

Check	Resultado esperado
11 tasks, cada una con TDD red→green→commit	✅
Tests verdes: `pytest packages/jw-meeting-scheduler/tests/` (~25 tests) + 2 CLI tests	✅
0 regresiones en suite global (2 716 tests baseline)	✅
Re-import idempotente	✅ (test_re_import_is_idempotent)
CRDT respect (newer local no sobrescrito)	✅ (test_upsert_respects_crdt_keeps_local_when_newer)
FieldEncryptor pattern (no Fernet raw)	✅ (Task 3 usa derive_key_from_password)
display_name_ciphered es `str`, no `bytes`	✅ (models.py + crypto.py)
Mapping real respeta PersonData.* envelope	✅ (Task 5 lee `.value` correctamente)
Gender derivado de male/female separados	✅ (test_map_person_unknown_gender_when_both_false)
Multi-congregación via JW_MEETING_SCHED_HOME	✅ (Task 7 _default_root + monkeypatch en tests)

Cómo verificar al cerrar

# 1. Sincronizar
uv sync --all-packages

# 2. Suite completa de F81.0
.venv/bin/python -m pytest packages/jw-meeting-scheduler/ packages/jw-cli/tests/test_scheduler_command.py -v

# 3. Ruff + mypy
.venv/bin/python -m ruff check packages/jw-meeting-scheduler/
.venv/bin/python -m ruff format --check packages/jw-meeting-scheduler/
.venv/bin/python -m mypy packages/jw-meeting-scheduler/src

# 4. Suite global (sin regresiones)
.venv/bin/python -m pytest

# 5. CLI smoke
uv run jw scheduler import \
  --backup packages/jw-meeting-scheduler/tests/fixtures/backup_minimal.json \
  --congregation smoke \
  --dry-run

Fase 81.0 — Importador organized-app — Plan de Implementación

Global Constraints

Task 1: Scaffold del package jw-meeting-scheduler

Task 2: Modelos Pydantic del scheduler

Task 3: Encryption helper sobre FieldEncryptor

Task 4: Loader del JSON backup organized-app

Task 5: Mapper PersonType → PersonRecord

Task 6: Mapper SchedWeek → AssignmentHistoryEntry[]

Task 7: Store SQLite + persistencia + idempotencia

Task 8: Dry-run + diff helpers

Task 9: Pipeline E2E run_import() + dry-run flag

Task 10: CLI command jw scheduler import

Task 11: Documentación operativa

Qué pasa por dentro

Ubicación del store

Cifrado

Re-import

Próximos pasos (F81.1+)

Self-review (al cerrar el plan)

Cómo verificar al cerrar

Fase 81.0 — Importador `organized-app` — Plan de Implementación

Task 1: Scaffold del package `jw-meeting-scheduler`

Task 3: Encryption helper sobre `FieldEncryptor`

Task 4: Loader del JSON backup `organized-app`

Task 5: Mapper `PersonType → PersonRecord`

Task 6: Mapper `SchedWeek → AssignmentHistoryEntry[]`

Task 9: Pipeline E2E `run_import()` + dry-run flag

Task 10: CLI command `jw scheduler import`