read.markets/docs/superpowers/plans/2026-05-27-localization-italian.md

Localization (Italian active, ES/FR/DE WIP) — Implementation Plan

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: Make every AI-generated user-facing surface render in Italian when the user picks Italiano in settings; lay the wiring so adding ES/FR/DE later is a one-line constant change.

Architecture: Per-user surfaces (portfolio_analysis, email_digest_job, follow-up chat if present) thread the user's lang into the LLM prompt via a respond_in_clause() helper — one extra line on the system prompt, no extra call. The hourly ai_log_job writes the English StrategicLog row as today, then fans out parallel translate() calls (asyncio.gather) — one per active non-en language with at least one user — and persists each result in a new strategic_log_translations table. The /log endpoint serves the matching translation when present and falls back to English otherwise.

Tech Stack: FastAPI · SQLAlchemy 2.0 async · Alembic · MariaDB (prod) / aiosqlite (tests) · existing openrouter.call_llm (DeepSeek-4-flash primary, OpenRouter fallback) · Jinja2 templates

Spec: docs/superpowers/specs/2026-05-27-localization-italian-design.md

---

File Structure

Create:

• app/services/i18n.py — LANGUAGES, ACTIVE_LANGUAGES, respond_in_clause()
• app/services/translation.py — translate(client, text, target_lang) wrapping call_llm
• alembic/versions/0022_localization.py — adds users.lang, creates strategic_log_translations
• tests/test_i18n.py — unit tests for the two new services
• tests/test_localization_integration.py — wiring/fan-out + route-level integration

Modify:

• app/models.py — add User.lang, new StrategicLogTranslation model
• app/jobs/ai_log_job.py — translation fan-out after English row is committed
• app/services/portfolio_analysis.py — accept + thread lang field
• app/routers/universe.py — pass cu.user.lang (or "en" for admin) into parse_request
• app/jobs/email_digest_job.py — thread user.lang into the per-user prompt
• app/routers/api.py — add PATCH /api/settings/language endpoint
• app/routers/pages.py — log_page / log_page_day serve translated content when available
• app/templates/settings.html — language dropdown + small JS handler

Reuse without modification:

• app/services/openrouter.call_llm, LogResult
• app/auth.require_auth / require_token / CurrentUser
• app/db.Base, utcnow, get_session
• Test session-factory pattern from tests/test_referral_conversion.py::_build_session_factory

---

Test Conventions

All tests runnable in the project-isolated container:

docker compose -f docker-compose.test.yml run --rm test pytest tests/test_i18n.py tests/test_localization_integration.py -v

DB-touching tests use the per-test _build_session_factory(tmp_path) pattern from tests/test_referral_conversion.py. LLM calls mocked via monkeypatch.setattr(<module>, "call_llm", AsyncMock(...)). Real Yahoo/network calls forbidden.

---

Task 1: i18n service — LANGUAGES, ACTIVE_LANGUAGES, respond_in_clause

Files:

• Create: app/services/i18n.py

• Test: tests/test_i18n.py

• Step 1: Write failing tests

Create tests/test_i18n.py:

"""Unit tests for app.services.i18n."""
from __future__ import annotations

import pytest


def test_languages_contains_all_four_plus_english():
    from app.services.i18n import LANGUAGES
    assert set(LANGUAGES.keys()) == {"en", "it", "es", "fr", "de"}
    assert LANGUAGES["en"] == "English"
    assert LANGUAGES["it"] == "Italian"
    assert LANGUAGES["es"] == "Spanish"
    assert LANGUAGES["fr"] == "French"
    assert LANGUAGES["de"] == "German"


def test_active_languages_is_en_and_it_only():
    from app.services.i18n import ACTIVE_LANGUAGES
    assert ACTIVE_LANGUAGES == {"en", "it"}


def test_respond_in_clause_empty_for_english():
    from app.services.i18n import respond_in_clause
    assert respond_in_clause("en") == ""


def test_respond_in_clause_empty_for_none_or_empty():
    from app.services.i18n import respond_in_clause
    assert respond_in_clause("") == ""
    assert respond_in_clause(None) == ""


def test_respond_in_clause_italian():
    from app.services.i18n import respond_in_clause
    result = respond_in_clause("it")
    assert "Italian" in result
    assert result.startswith("\n\n")


def test_respond_in_clause_unknown_lang_falls_back_to_english():
    """Defensive: a raw POST or stale lang code should not crash the
    prompt assembly. Unknown codes map to no-suffix (English default)."""
    from app.services.i18n import respond_in_clause
    assert respond_in_clause("xx") == ""

• Step 2: Run tests to verify they fail

docker compose -f docker-compose.test.yml run --rm test pytest tests/test_i18n.py -v

Expected: 6 FAIL with ImportError.

• Step 3: Implement app/services/i18n.py

"""Language registry + prompt helpers for localized AI output.

Two surfaces consume this module:
- Per-user LLM call sites (portfolio analysis, digest, chat) call
  ``respond_in_clause(user.lang)`` and append the result to their
  system prompt.
- The settings dropdown + its PATCH endpoint consult ``ACTIVE_LANGUAGES``
  to decide which options are selectable.

Adding Spanish/French/German support later is a one-line constant
change: extend ``ACTIVE_LANGUAGES`` to include the new code. No other
code change is required — the rest of the system already treats them
as first-class via ``LANGUAGES``.
"""
from __future__ import annotations


# Display labels for every language the system knows about. ES/FR/DE
# are kept here so labels still render in the dropdown (as disabled
# options) without requiring code changes to enable them later.
LANGUAGES: dict[str, str] = {
    "en": "English",
    "it": "Italian",
    "es": "Spanish",
    "fr": "French",
    "de": "German",
}


# Languages users can actually select. Settings POST validates against
# this; the strategic-log translation fan-out only considers these.
ACTIVE_LANGUAGES: set[str] = {"en", "it"}


def respond_in_clause(lang: str | None) -> str:
    """Suffix appended to per-user LLM system prompts.

    Returns an empty string for ``en`` (no nudge needed), an unknown
    code, or ``None``/empty input — those callers want the default
    English path. Otherwise returns ``"\\n\\nRespond in <Language>."``
    keyed off ``LANGUAGES``.
    """
    if not lang or lang == "en" or lang not in LANGUAGES:
        return ""
    return f"\n\nRespond in {LANGUAGES[lang]}."

• Step 4: Run tests to verify they pass

docker compose -f docker-compose.test.yml run --rm test pytest tests/test_i18n.py -v

Expected: 6 PASS.

• Step 5: Commit

git add app/services/i18n.py tests/test_i18n.py
git commit -m "i18n: add LANGUAGES, ACTIVE_LANGUAGES, respond_in_clause helper"

Context

• Working directory: /home/gg/mydocker_images/products/read.markets. Branch main. Commit directly.
• Test runner: ONLY docker compose -f docker-compose.test.yml .... NEVER plain docker compose ... against the prod stack.
• If git commit is blocked by the auto-mode classifier, leave the tree dirty and report — the controller will commit.

---

Task 2: translation service

Files:

• Create: app/services/translation.py

• Test: tests/test_i18n.py (append)

• Step 1: Write failing tests

Append to tests/test_i18n.py:

@pytest.mark.asyncio
async def test_translate_happy_path(monkeypatch):
    from unittest.mock import AsyncMock, MagicMock

    from app.services import translation as mod
    from app.services.openrouter import LogResult

    monkeypatch.setattr(mod, "call_llm", AsyncMock(return_value=LogResult(
        content="# Apertura\n\nIl mercato è in calo dello 0,4%.",
        model="deepseek/deepseek-v4-flash",
        prompt_tokens=300, completion_tokens=80, cost_usd=0.00002,
    )))

    client = MagicMock()
    translated, llm_log = await mod.translate(
        client, "# Open\n\nThe market is down 0.4%.", "it",
    )
    assert "Apertura" in translated
    assert llm_log.model == "deepseek/deepseek-v4-flash"
    assert llm_log.cost_usd == pytest.approx(0.00002)


@pytest.mark.asyncio
async def test_translate_strips_code_fences(monkeypatch):
    """If the LLM wraps the output in ```markdown ... ```, strip it."""
    from unittest.mock import AsyncMock, MagicMock

    from app.services import translation as mod
    from app.services.openrouter import LogResult

    fenced = "```markdown\n# Titolo\n\nCorpo.\n```"
    monkeypatch.setattr(mod, "call_llm", AsyncMock(return_value=LogResult(
        content=fenced, model="m", prompt_tokens=10, completion_tokens=20, cost_usd=0.0,
    )))

    client = MagicMock()
    translated, _ = await mod.translate(client, "# Title\n\nBody.", "it")
    assert "```" not in translated
    assert translated.startswith("# Titolo")


@pytest.mark.asyncio
async def test_translate_provider_failure_propagates(monkeypatch):
    from unittest.mock import AsyncMock, MagicMock

    from app.services import translation as mod

    monkeypatch.setattr(mod, "call_llm", AsyncMock(side_effect=RuntimeError("upstream down")))

    client = MagicMock()
    with pytest.raises(RuntimeError, match="upstream down"):
        await mod.translate(client, "# Title\n\nBody.", "it")


@pytest.mark.asyncio
async def test_translate_unknown_lang_returns_source_unchanged(monkeypatch):
    """Defensive: an unknown lang code (or 'en') short-circuits without
    calling the LLM. Callers shouldn't have to gate the call themselves."""
    from unittest.mock import AsyncMock, MagicMock

    from app.services import translation as mod
    from app.services.openrouter import LogResult

    call_mock = AsyncMock(return_value=LogResult(
        content="should not be returned",
        model="m", prompt_tokens=0, completion_tokens=0, cost_usd=0.0,
    ))
    monkeypatch.setattr(mod, "call_llm", call_mock)

    client = MagicMock()
    out, _ = await mod.translate(client, "Hello world.", "en")
    assert out == "Hello world."
    call_mock.assert_not_awaited()

• Step 2: Run tests to verify they fail

docker compose -f docker-compose.test.yml run --rm test pytest tests/test_i18n.py -k translate -v

Expected: 4 FAIL with ImportError.

• Step 3: Implement app/services/translation.py

"""Markdown translation via the existing LLM provider chain.

DeepSeek-4-flash at ~$0.28/M output tokens is cheap enough that we
don't bother with a separate translation-only model. ``call_llm``'s
provider chain (DeepSeek primary, OpenRouter fallback) handles this
path identically to any other LLM call.

The translator is content-aware in one important way: it instructs the
model to preserve markdown structure, ticker symbols, numbers, dates,
and percentages verbatim. This keeps generated artefacts (tables of
quotes, embedded percentages, dated references) intact across the
translation boundary.
"""
from __future__ import annotations

import httpx

from app.services.i18n import LANGUAGES
from app.services.openrouter import LogResult, call_llm


_SYSTEM_PROMPT_TMPL = """\
You are an expert translator working on financial-markets commentary.
Translate the following markdown text to {language}.

Strict rules:
- Preserve ALL markdown formatting (headings, lists, emphasis, links,
  tables, code spans).
- Do NOT translate ticker symbols (AAPL, MSFT, VOD.L, ASML.AS, etc.),
  company legal names, percentages, dates, ISO currency codes, or any
  numbers.
- Do NOT add commentary, preambles, or apologies. Output ONLY the
  translated markdown.
"""


async def translate(
    client: httpx.AsyncClient,
    text: str,
    target_lang: str,
) -> tuple[str, LogResult]:
    """Translate markdown ``text`` to ``target_lang``.

    Returns ``(translated_markdown, LogResult)``. Caller persists the
    cost/model provenance from LogResult next to the cached row.

    Short-circuits without calling the LLM when ``target_lang`` is
    ``'en'``, unknown, or empty — returns the source unchanged with a
    zero-cost stub LogResult. This lets fan-out callers iterate over
    all languages without per-call gating.

    Raises on provider failure (HTTP error, all chain providers down).
    Callers in fan-out paths should catch and log per-language.
    """
    if not target_lang or target_lang == "en" or target_lang not in LANGUAGES:
        # No-op fast path. Returning a fake LogResult keeps the call
        # signature stable for callers who unpack the tuple.
        return text, LogResult(
            content=text, model="noop",
            prompt_tokens=0, completion_tokens=0, cost_usd=0.0,
        )

    system_prompt = _SYSTEM_PROMPT_TMPL.format(language=LANGUAGES[target_lang])
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user",   "content": text},
    ]
    result = await call_llm(client, messages)

    content = (result.content or "").strip()
    # Strip code fences if the model wrapped its output despite the system rule.
    if content.startswith("```"):
        # Drop the opening fence (with optional language tag).
        first_nl = content.find("\n")
        if first_nl != -1:
            content = content[first_nl + 1:]
        # Drop the closing fence.
        if content.rstrip().endswith("```"):
            content = content.rstrip()[:-3].rstrip()
        content = content.strip()

    return content, result

• Step 4: Run tests to verify they pass

docker compose -f docker-compose.test.yml run --rm test pytest tests/test_i18n.py -v

Expected: 10 tests pass total (6 from Task 1 + 4 new).

• Step 5: Commit

git add app/services/translation.py tests/test_i18n.py
git commit -m "i18n: add translate() helper backed by call_llm"

---

Task 3: User.lang column + StrategicLogTranslation model

Files:

• Modify: app/models.py

• Test: tests/test_localization_integration.py

• Step 1: Write failing tests

Create tests/test_localization_integration.py:

"""Integration tests: model surface, ai_log_job translation fan-out,
route-level localized fetch, settings PATCH validation."""
from __future__ import annotations

import pytest


def _build_session_factory(tmp_path):
    """Per-test sqlite engine + factory. Mirrors test_referral_conversion.py."""
    from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

    from app import db as db_mod
    from app.db import Base
    import app.models  # noqa: F401 — registers models on Base.metadata

    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/loc.db")
    factory = async_sessionmaker(engine, expire_on_commit=False)
    db_mod._engine = engine
    db_mod._session_factory = factory

    async def _setup():
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)

    return engine, factory, _setup


def test_user_has_lang_column_with_default_en():
    from sqlalchemy import inspect
    from app.models import User

    cols = {c.name: c for c in inspect(User).columns}
    assert "lang" in cols
    assert cols["lang"].nullable is False
    # SQLAlchemy default may be a callable or a literal — check both.
    default = cols["lang"].default
    assert default is not None
    if hasattr(default, "arg"):
        assert default.arg == "en"


def test_strategic_log_translation_model_columns():
    from sqlalchemy import inspect
    from app.models import StrategicLogTranslation

    cols = {c.name: c for c in inspect(StrategicLogTranslation).columns}
    assert "log_id" in cols
    assert "lang" in cols
    assert "content_md" in cols
    assert "generated_at" in cols
    assert "llm_model" in cols
    assert "llm_cost_usd" in cols
    assert cols["log_id"].nullable is False
    assert cols["lang"].nullable is False
    assert cols["content_md"].nullable is False

• Step 2: Run tests to verify they fail

docker compose -f docker-compose.test.yml run --rm test pytest tests/test_localization_integration.py -v

Expected: 2 FAIL — first on User.lang missing, second on StrategicLogTranslation import error.

• Step 3: Add User.lang column

In app/models.py, find the User class. Find a sensible place near other user-preference columns (next to tone or digest_tone) and add:

    # Preferred language for AI-generated content (strategic log,
    # digest emails, portfolio commentary). Default 'en'. The settings
    # PATCH endpoint validates against ACTIVE_LANGUAGES in
    # app/services/i18n.py before writing.
    lang: Mapped[str] = mapped_column(
        String(8), nullable=False, default="en", server_default="en",
    )

• Step 4: Add StrategicLogTranslation model

In app/models.py, append after the existing StrategicLog class (around line 108-122):

class StrategicLogTranslation(Base):
    """Cached translation of a single StrategicLog row.

    Populated by ai_log_job after the English row is committed: one
    row per (log_id, lang) combination. The /log endpoint serves the
    matching row when available and falls back to the English source
    when no row exists yet (e.g. translation failed or the language
    was added after the log was generated).

    No user attribution — the cache is shared. Setting `lang` on a
    user just selects which (already-translated) variant they see.
    """
    __tablename__ = "strategic_log_translations"

    id: Mapped[int] = mapped_column(_PK, primary_key=True, autoincrement=True)
    log_id: Mapped[int] = mapped_column(
        _PK, ForeignKey("strategic_logs.id", ondelete="CASCADE"), nullable=False,
    )
    lang: Mapped[str] = mapped_column(String(8), nullable=False)
    content_md: Mapped[str] = mapped_column(Text, nullable=False)
    generated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, default=utcnow,
    )
    llm_model: Mapped[str | None] = mapped_column(String(64))
    llm_cost_usd: Mapped[float | None] = mapped_column(Float)

    __table_args__ = (
        UniqueConstraint("log_id", "lang", name="uq_slt_log_lang"),
    )

• Step 5: Run tests to verify they pass

docker compose -f docker-compose.test.yml run --rm test pytest tests/test_localization_integration.py -v

Expected: 2 PASS.

• Step 6: Commit

git add app/models.py tests/test_localization_integration.py
git commit -m "models: add User.lang + StrategicLogTranslation"

Context for this task

• The existing User class already imports String, mapped_column, Mapped. No new imports needed for User.lang.
• For StrategicLogTranslation, _PK, String, Text, DateTime, Float, ForeignKey, UniqueConstraint, Mapped, mapped_column, Base, utcnow, datetime are all already imported at the top of app/models.py — no new imports needed.

---

Task 4: Alembic migration 0022

Files:

• Create: alembic/versions/0022_localization.py

• Step 1: Write the migration

"""localization: users.lang + strategic_log_translations.

Revision ID: 0022
Revises: 0021
Create Date: 2026-05-27
"""
from typing import Sequence, Union

import sqlalchemy as sa
from alembic import op


revision: str = "0022"
down_revision: Union[str, None] = "0021"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    op.add_column(
        "users",
        sa.Column(
            "lang", sa.String(length=8), nullable=False,
            server_default="en",
        ),
    )
    op.create_table(
        "strategic_log_translations",
        sa.Column("id", sa.BigInteger(), primary_key=True, autoincrement=True),
        sa.Column("log_id", sa.BigInteger(), nullable=False),
        sa.Column("lang", sa.String(length=8), nullable=False),
        sa.Column("content_md", sa.Text(), nullable=False),
        sa.Column("generated_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("llm_model", sa.String(length=64), nullable=True),
        sa.Column("llm_cost_usd", sa.Float(), nullable=True),
        sa.ForeignKeyConstraint(
            ["log_id"], ["strategic_logs.id"],
            ondelete="CASCADE", name="fk_slt_log",
        ),
        sa.UniqueConstraint("log_id", "lang", name="uq_slt_log_lang"),
    )


def downgrade() -> None:
    op.drop_table("strategic_log_translations")
    op.drop_column("users", "lang")

• Step 2: Verify migration chain integrity

docker compose -f docker-compose.test.yml run --rm test python -c "
from alembic.config import Config
from alembic.script import ScriptDirectory
sd = ScriptDirectory.from_config(Config('alembic.ini'))
heads = sd.get_heads()
assert heads == ('0022',), heads
rev = sd.get_revision('0022')
assert rev.down_revision == '0021'
print('OK')
"

Expected: prints OK.

• Step 3: Commit

git add alembic/versions/0022_localization.py
git commit -m "alembic: add 0022 localization (users.lang + strategic_log_translations)"

Context

• Previous migration is 0021_csv_format_template.py. The new file follows the same hand-rolled style.
• Codebase convention for integer server_defaults is sa.text("0") — but here we have no integer defaults to set. String default "en" uses the bare string per existing migrations (e.g. 0011_drop_portfolio_tables.py uses server_default="GBP").

---

Task 5: ai_log_job translation fan-out

Files:

• Modify: app/jobs/ai_log_job.py

• Test: tests/test_localization_integration.py (append)

• Step 1: Inspect the existing log-writing path

grep -n "def \|StrategicLog\|session.commit\|session.add" app/jobs/ai_log_job.py | head -20

Locate the function and the line where the new StrategicLog row is committed. The fan-out runs after that commit.

• Step 2: Write failing tests

Append to tests/test_localization_integration.py:

@pytest.mark.asyncio
async def test_log_translation_fanout_no_active_non_en_users(tmp_path, monkeypatch):
    """When no users have an active non-en lang, the fan-out makes no
    translation calls and no rows are inserted."""
    from unittest.mock import AsyncMock
    from sqlalchemy import select

    from app.db import utcnow
    from app.models import StrategicLog, StrategicLogTranslation, User
    from app.jobs import ai_log_job

    _, factory, setup = _build_session_factory(tmp_path)
    await setup()

    fake_translate = AsyncMock()
    monkeypatch.setattr(ai_log_job, "translate", fake_translate)

    # Seed an English user (no non-en users).
    async with factory() as session:
        session.add(User(id=1, email="en@x", tier="paid", lang="en"))
        slog = StrategicLog(
            generated_at=utcnow(), content_md="# Open\n\nDown 0.4%.",
            tone="INTERMEDIATE", analysis="NORMAL",
        )
        session.add(slog)
        await session.commit()
        log_id = slog.id

    async with factory() as session:
        await ai_log_job.translate_log_for_active_languages(session, log_id)

    fake_translate.assert_not_awaited()
    async with factory() as session:
        rows = (await session.execute(select(StrategicLogTranslation))).scalars().all()
    assert rows == []


@pytest.mark.asyncio
async def test_log_translation_fanout_italian_user(tmp_path, monkeypatch):
    """One user at lang=it triggers one translation; the row lands with
    the right lang and log_id."""
    from unittest.mock import AsyncMock
    from sqlalchemy import select

    from app.db import utcnow
    from app.models import StrategicLog, StrategicLogTranslation, User
    from app.services.openrouter import LogResult
    from app.jobs import ai_log_job

    _, factory, setup = _build_session_factory(tmp_path)
    await setup()

    async def _fake_translate(client, text, target_lang):
        assert target_lang == "it"
        return "# Apertura\n\nIn calo 0,4%.", LogResult(
            content="# Apertura\n\nIn calo 0,4%.",
            model="deepseek/deepseek-v4-flash",
            prompt_tokens=300, completion_tokens=80, cost_usd=0.00002,
        )
    monkeypatch.setattr(ai_log_job, "translate", _fake_translate)

    async with factory() as session:
        session.add(User(id=2, email="it@x", tier="paid", lang="it"))
        slog = StrategicLog(
            generated_at=utcnow(), content_md="# Open\n\nDown 0.4%.",
            tone="INTERMEDIATE", analysis="NORMAL",
        )
        session.add(slog)
        await session.commit()
        log_id = slog.id

    async with factory() as session:
        await ai_log_job.translate_log_for_active_languages(session, log_id)

    async with factory() as session:
        rows = (await session.execute(select(StrategicLogTranslation))).scalars().all()
    assert len(rows) == 1
    row = rows[0]
    assert row.log_id == log_id
    assert row.lang == "it"
    assert row.content_md.startswith("# Apertura")
    assert row.llm_model == "deepseek/deepseek-v4-flash"
    assert row.llm_cost_usd == pytest.approx(0.00002)


@pytest.mark.asyncio
async def test_log_translation_fanout_per_language_failure_isolated(tmp_path, monkeypatch):
    """If one language's translation fails, the others (if any) still land
    and the job does not raise."""
    from unittest.mock import AsyncMock
    from sqlalchemy import select

    from app.db import utcnow
    from app.models import StrategicLog, StrategicLogTranslation, User
    from app.jobs import ai_log_job

    _, factory, setup = _build_session_factory(tmp_path)
    await setup()

    async def _fake_translate(client, text, target_lang):
        raise RuntimeError("upstream down")
    monkeypatch.setattr(ai_log_job, "translate", _fake_translate)

    async with factory() as session:
        session.add(User(id=3, email="it@x", tier="paid", lang="it"))
        slog = StrategicLog(
            generated_at=utcnow(), content_md="# Open",
            tone="INTERMEDIATE", analysis="NORMAL",
        )
        session.add(slog)
        await session.commit()
        log_id = slog.id

    # Must NOT raise.
    async with factory() as session:
        await ai_log_job.translate_log_for_active_languages(session, log_id)

    async with factory() as session:
        rows = (await session.execute(select(StrategicLogTranslation))).scalars().all()
    assert rows == []

• Step 3: Run tests to verify they fail

docker compose -f docker-compose.test.yml run --rm test pytest tests/test_localization_integration.py -k log_translation_fanout -v

Expected: 3 FAIL with AttributeError: module 'app.jobs.ai_log_job' has no attribute 'translate_log_for_active_languages'.

• Step 4: Add the fan-out function

In app/jobs/ai_log_job.py, add (at module scope, alongside other helpers; use existing imports + add what's missing — httpx, asyncio, select, the i18n + translation modules, and the model imports):

import asyncio
import httpx

from sqlalchemy import select

from app.db import utcnow
from app.models import User, StrategicLogTranslation
from app.services.i18n import ACTIVE_LANGUAGES
from app.services.translation import translate

(Add only the lines not already present.)

Then add the function:

async def translate_log_for_active_languages(session, log_id: int) -> None:
    """Fan out per-language translations for the strategic log identified
    by ``log_id``.

    Reads ``users.lang`` (deduplicated, restricted to ACTIVE_LANGUAGES
    minus English), one translation call per language in parallel via
    ``asyncio.gather``, persists each successful result as a
    ``StrategicLogTranslation`` row. Per-language failures are logged
    but never raise — the strategic log itself is already committed at
    this point and translation is a best-effort enhancement.

    The job orchestrator calls this AFTER the English ``StrategicLog``
    row is committed; pass the row's ``id`` in.
    """
    from app.models import StrategicLog  # local import: avoid widening top-level imports
    target_langs = sorted({l for l in ACTIVE_LANGUAGES if l != "en"})
    if not target_langs:
        return

    active_langs = (await session.execute(
        select(User.lang).distinct().where(User.lang.in_(target_langs))
    )).scalars().all()
    if not active_langs:
        return

    log_row = await session.get(StrategicLog, log_id)
    if log_row is None:
        log.warning("log.translate.missing_log", log_id=log_id)
        return

    async with httpx.AsyncClient(follow_redirects=True, timeout=60) as client:
        results = await asyncio.gather(*[
            translate(client, log_row.content_md, lang)
            for lang in active_langs
        ], return_exceptions=True)

    for lang, result in zip(active_langs, results):
        if isinstance(result, Exception):
            log.warning("log.translate.failed", lang=lang, log_id=log_id,
                        error=str(result)[:200])
            continue
        translated_md, llm_log = result
        session.add(StrategicLogTranslation(
            log_id=log_id, lang=lang,
            content_md=translated_md,
            generated_at=utcnow(),
            llm_model=llm_log.model,
            llm_cost_usd=llm_log.cost_usd,
        ))
    await session.commit()

• Step 5: Wire the fan-out into the existing log-write path

Find the function in ai_log_job.py that writes a StrategicLog row and calls session.commit(). After that commit, capture the row's id and call:

await translate_log_for_active_languages(session, slog.id)

(Use whatever the actual local variable name for the row is.)

• Step 6: Run tests to verify they pass

docker compose -f docker-compose.test.yml run --rm test pytest tests/test_localization_integration.py -v

Expected: 5 tests pass (2 from Task 3 + 3 new).

• Step 7: Commit

git add app/jobs/ai_log_job.py tests/test_localization_integration.py
git commit -m "ai-log-job: translate strategic log for active non-en languages"

---

Task 6: portfolio_analysis localization

Files:

• Modify: app/services/portfolio_analysis.py

• Modify: app/routers/universe.py — the /api/analyze route

• Test: tests/test_localization_integration.py (append)

• Step 1: Write failing tests

Append to tests/test_localization_integration.py:

@pytest.mark.asyncio
async def test_analyse_threads_lang_into_system_prompt(monkeypatch):
    """When lang='it', the system prompt sent to call_llm contains
    'Respond in Italian.' — the LLM does the rest."""
    from unittest.mock import AsyncMock
    from app.services import portfolio_analysis as pa
    from app.services.openrouter import LogResult

    captured = {}

    async def _fake_call_llm(client, messages, **kw):
        captured["messages"] = messages
        return LogResult(
            content="Analisi del portafoglio in italiano.",
            model="m", prompt_tokens=400, completion_tokens=100, cost_usd=0.0001,
        )
    monkeypatch.setattr(pa, "call_llm", _fake_call_llm)

    payload = {
        "positions": [{"yahoo_ticker": "AAPL", "qty": 10, "avg_cost": 150.0,
                       "currency": "USD", "name": "Apple Inc"}],
        "prices": {"AAPL": {"p": 172.4, "c": "USD"}},
        "fx": {"USD": 1.0},
        "base_currency": "USD",
        "tone": "INTERMEDIATE",
        "analysis": "NORMAL",
        "lang": "it",
    }
    req = pa.parse_request(payload)
    assert req.lang == "it"

    # Direct call into analyse() to inspect the captured prompt.
    # Use None session — analyse should not touch the DB in this code path
    # because we mocked call_llm before any AICall ledger write.
    # If analyse insists on a session, wrap with the test factory.
    result = await pa.analyse(None, req)  # noqa: ARG  — session ignored by mock
    system = next(m["content"] for m in captured["messages"] if m["role"] == "system")
    assert "Respond in Italian" in system


@pytest.mark.asyncio
async def test_analyse_no_clause_when_lang_is_en(monkeypatch):
    from unittest.mock import AsyncMock
    from app.services import portfolio_analysis as pa
    from app.services.openrouter import LogResult

    captured = {}

    async def _fake_call_llm(client, messages, **kw):
        captured["messages"] = messages
        return LogResult(
            content="Portfolio analysis in English.",
            model="m", prompt_tokens=400, completion_tokens=100, cost_usd=0.0001,
        )
    monkeypatch.setattr(pa, "call_llm", _fake_call_llm)

    payload = {
        "positions": [{"yahoo_ticker": "AAPL", "qty": 10, "avg_cost": 150.0,
                       "currency": "USD", "name": "Apple Inc"}],
        "prices": {"AAPL": {"p": 172.4, "c": "USD"}},
        "fx": {"USD": 1.0},
        "base_currency": "USD",
        "tone": "INTERMEDIATE",
        "analysis": "NORMAL",
        "lang": "en",
    }
    req = pa.parse_request(payload)
    await pa.analyse(None, req)
    system = next(m["content"] for m in captured["messages"] if m["role"] == "system")
    assert "Respond in" not in system

• Step 2: Run tests to verify they fail

docker compose -f docker-compose.test.yml run --rm test pytest tests/test_localization_integration.py -k analyse -v

Expected: 2 FAIL — either lang field on AnalysisRequest missing, or "Respond in Italian" not in the prompt.

• Step 3: Add lang to AnalysisRequest and thread through

In app/services/portfolio_analysis.py:

1. Locate the AnalysisRequest dataclass (or pydantic model). Add a lang: str = "en" field next to tone / analysis.
2. Locate parse_request. Read payload.get("lang", "en") and pass it to the request constructor. Validate against LANGUAGES:

from app.services.i18n import LANGUAGES, respond_in_clause

# Inside parse_request, alongside the existing tone/analysis parsing:
lang = (payload.get("lang") or "en").strip().lower()
if lang not in LANGUAGES:
    lang = "en"

Then build the request with lang=lang.

3. Locate analyse (the async function that builds messages and calls call_llm). After the system prompt is composed, append the i18n clause:

system_prompt = system_prompt + respond_in_clause(req.lang)

(Use whatever the local variable name for the system prompt is.)

• Step 4: Pass the user's lang from the route

In app/routers/universe.py, find analyze_portfolio (the /api/analyze route handler). Add the user's lang to the payload before calling parse_request:

# Just before parse_request:
user_lang = (
    principal.user.lang if (principal.user and principal.user.lang) else "en"
)
payload["lang"] = user_lang

(The handler receives principal via Depends. Confirm by reading the handler's signature; if the principal isn't already wired in, add principal: CurrentUser = Depends(require_paid) matching the existing dep.)

• Step 5: Run tests to verify they pass

docker compose -f docker-compose.test.yml run --rm test pytest tests/test_localization_integration.py -v

Expected: 7 tests pass (5 from Tasks 3+5 + 2 new).

• Step 6: Commit

git add app/services/portfolio_analysis.py app/routers/universe.py tests/test_localization_integration.py
git commit -m "analyse: thread user.lang into the system prompt"

---

Task 7: email_digest_job localization

Files:

• Modify: app/jobs/email_digest_job.py

• Test: tests/test_localization_integration.py (append)

• Step 1: Write failing test

Append to tests/test_localization_integration.py:

@pytest.mark.asyncio
async def test_digest_threads_lang_into_system_prompt(monkeypatch):
    """The per-user digest generation appends 'Respond in Italian.' to
    the system prompt when the user is Italian."""
    from unittest.mock import AsyncMock
    from app.jobs import email_digest_job as ed
    from app.services.openrouter import LogResult

    captured = []

    async def _fake_call_llm(client, messages, **kw):
        captured.append(messages)
        return LogResult(
            content="**Apertura.** Il mercato è in calo.",
            model="m", prompt_tokens=300, completion_tokens=400, cost_usd=0.0001,
        )
    monkeypatch.setattr(ed, "call_llm", _fake_call_llm)

    # _generate_variants is the helper that runs one LLM call per tone.
    # It takes a context dict and a 'kind' (daily/weekly). The exact
    # signature is in app/jobs/email_digest_job.py — inspect before
    # calling. The test below assumes it accepts a `target_lang` kwarg.
    from datetime import datetime, timezone

    ctx = {
        "today": datetime.now(timezone.utc),
        "quotes_by_group": {},
        "headlines_by_bucket": {},
        "reference_line": None,
    }

    # `_generate_variants` should iterate tones internally; we just need
    # to assert at least one captured system prompt has the IT clause.
    import httpx
    async with httpx.AsyncClient() as client:
        await ed._generate_variants(None, client, "daily", ctx, target_lang="it")

    assert captured, "no LLM call was made"
    italian_found = any(
        any(
            m["role"] == "system" and "Respond in Italian" in m["content"]
            for m in messages
        )
        for messages in captured
    )
    assert italian_found, "no system prompt contained 'Respond in Italian'"

• Step 2: Run test to verify it fails

docker compose -f docker-compose.test.yml run --rm test pytest tests/test_localization_integration.py -k digest_threads_lang -v

Expected: FAIL — either _generate_variants doesn't accept target_lang, or the IT clause isn't in the prompt.

• Step 3: Thread target_lang through _generate_variants and the per-user driver

In app/jobs/email_digest_job.py:

1. Import the helper:

   from app.services.i18n import respond_in_clause
   

2. Find _generate_variants. Add target_lang: str = "en" to its signature. Where it composes each variant's system prompt, append:

   system_prompt = system_prompt + respond_in_clause(target_lang)
   

3. Find the per-user send path (the function that actually iterates users — likely _send_for_user or similar, called from the job's main loop). Where it calls _generate_variants, pass target_lang=user.lang:

   variants = await _generate_variants(
       session, client, kind, ctx, target_lang=user.lang,
   )
   

   If the existing call site is in the main job loop and constructs variants once for all users, that breaks the "per-user language" contract. In that case the variants must be generated PER USER, not globally. Look for the caller; if it caches variants across users, restructure to call _generate_variants inside the per-user loop. Important: if this requires more than a few lines of change, stop and report a concern — the existing assumption may be wrong and we want explicit guidance.

• Step 4: Run tests to verify they pass

docker compose -f docker-compose.test.yml run --rm test pytest tests/test_localization_integration.py -v

Expected: all tests pass (8 total now).

• Step 5: Commit

git add app/jobs/email_digest_job.py tests/test_localization_integration.py
git commit -m "digest: thread user.lang into per-user generation"

---

Task 8: /log endpoint localized fetch

Files:

• Modify: app/routers/pages.py — log_page and log_page_day

• Test: tests/test_localization_integration.py (append)

• Step 1: Inspect the existing log endpoints

grep -n "def log_page\|StrategicLog\|content_md\|generated_at" app/routers/pages.py | head -20

Locate the function(s) that fetch the strategic log and pass content_md to the template.

• Step 2: Write a failing test

Append to tests/test_localization_integration.py:

@pytest.mark.asyncio
async def test_log_endpoint_serves_italian_when_user_is_italian(tmp_path):
    """When a user with lang='it' opens /log, the served content_md is
    the Italian translation, not the English original."""
    from datetime import datetime, timezone

    from app.db import utcnow
    from app.models import StrategicLog, StrategicLogTranslation, User

    _, factory, setup = _build_session_factory(tmp_path)
    await setup()

    async with factory() as session:
        session.add(User(id=10, email="it@x", tier="paid", lang="it"))
        slog = StrategicLog(
            generated_at=utcnow(), content_md="# Open\n\nDown 0.4%.",
            tone="INTERMEDIATE", analysis="NORMAL",
        )
        session.add(slog)
        await session.commit()
        session.add(StrategicLogTranslation(
            log_id=slog.id, lang="it",
            content_md="# Apertura\n\nIn calo 0,4%.",
            generated_at=utcnow(), llm_model="m", llm_cost_usd=0.0,
        ))
        await session.commit()
        log_id = slog.id

    # We test the resolver function directly rather than spinning up the
    # FastAPI TestClient — the resolver shape returns the rendered MD.
    from app.routers.pages import _resolve_log_content
    async with factory() as session:
        user = await session.get(User, 10)
        content = await _resolve_log_content(session, log_id, user.lang)
    assert "Apertura" in content
    assert "Open" not in content


@pytest.mark.asyncio
async def test_log_endpoint_falls_back_to_english_when_no_translation(tmp_path):
    """User lang='it' but no IT translation exists → English fallback."""
    from app.db import utcnow
    from app.models import StrategicLog, User

    _, factory, setup = _build_session_factory(tmp_path)
    await setup()

    async with factory() as session:
        session.add(User(id=11, email="it2@x", tier="paid", lang="it"))
        slog = StrategicLog(
            generated_at=utcnow(), content_md="# Open\n\nDown 0.4%.",
            tone="INTERMEDIATE", analysis="NORMAL",
        )
        session.add(slog)
        await session.commit()
        log_id = slog.id

    from app.routers.pages import _resolve_log_content
    async with factory() as session:
        user = await session.get(User, 11)
        content = await _resolve_log_content(session, log_id, user.lang)
    assert "Open" in content

• Step 3: Run tests to verify they fail

docker compose -f docker-compose.test.yml run --rm test pytest tests/test_localization_integration.py -k log_endpoint -v

Expected: 2 FAIL — _resolve_log_content doesn't exist yet.

• Step 4: Add the resolver and wire it into log_page / log_page_day

In app/routers/pages.py, add the resolver as a module-level async function:

async def _resolve_log_content(
    session: AsyncSession, log_id: int, lang: str,
) -> str:
    """Return the markdown content of strategic log ``log_id`` in the
    user's preferred language.

    If ``lang`` is ``en`` or no translation exists for the requested
    language, returns the English original. The fallback is silent —
    a missing translation is the expected case for hours where
    translation hasn't yet run."""
    from app.models import StrategicLog, StrategicLogTranslation

    if lang and lang != "en":
        row = (await session.execute(
            select(StrategicLogTranslation)
            .where(StrategicLogTranslation.log_id == log_id)
            .where(StrategicLogTranslation.lang == lang)
        )).scalar_one_or_none()
        if row is not None:
            return row.content_md
    log_row = await session.get(StrategicLog, log_id)
    return log_row.content_md if log_row is not None else ""

Then in log_page (and log_page_day if present), replace the line that pulls content_md directly from the StrategicLog row with a call to _resolve_log_content(session, log.id, cu.user.lang if cu.user else "en").

• Step 5: Run tests to verify they pass

docker compose -f docker-compose.test.yml run --rm test pytest tests/test_localization_integration.py -v

Expected: all 10 tests pass.

• Step 6: Commit

git add app/routers/pages.py tests/test_localization_integration.py
git commit -m "log: serve translated content when available; English fallback"

---

Task 9: PATCH /api/settings/language endpoint

Files:

• Modify: app/routers/api.py

• Test: tests/test_localization_integration.py (append)

• Step 1: Write failing tests

Append to tests/test_localization_integration.py:

@pytest.mark.asyncio
async def test_patch_language_accepts_active(tmp_path):
    """PATCH /api/settings/language accepts 'en' and 'it' and persists."""
    from app.models import User
    from app.routers.api import patch_language_prefs, LanguagePrefsIn
    from app.auth import CurrentUser

    _, factory, setup = _build_session_factory(tmp_path)
    await setup()

    async with factory() as session:
        session.add(User(id=20, email="u@x", tier="paid", lang="en"))
        await session.commit()

    class _P:
        is_admin = False
        def __init__(self, u): self.user = u

    async with factory() as session:
        user = await session.get(User, 20)
        result = await patch_language_prefs(
            payload=LanguagePrefsIn(lang="it"),
            principal=_P(user),
            session=session,
        )
        assert result.lang == "it"

    async with factory() as session:
        user = await session.get(User, 20)
    assert user.lang == "it"


@pytest.mark.asyncio
async def test_patch_language_rejects_wip(tmp_path):
    """PATCH rejects 'es'/'fr'/'de'/'xx' with 400 — ACTIVE_LANGUAGES gate."""
    from fastapi import HTTPException
    from app.models import User
    from app.routers.api import patch_language_prefs, LanguagePrefsIn

    _, factory, setup = _build_session_factory(tmp_path)
    await setup()

    async with factory() as session:
        session.add(User(id=21, email="u2@x", tier="paid", lang="en"))
        await session.commit()

    class _P:
        is_admin = False
        def __init__(self, u): self.user = u

    for bad in ("es", "fr", "de", "xx"):
        async with factory() as session:
            user = await session.get(User, 21)
            with pytest.raises(HTTPException) as exc:
                await patch_language_prefs(
                    payload=LanguagePrefsIn(lang=bad),
                    principal=_P(user),
                    session=session,
                )
            assert exc.value.status_code == 400

• Step 2: Run tests to verify they fail

docker compose -f docker-compose.test.yml run --rm test pytest tests/test_localization_integration.py -k patch_language -v

Expected: 2 FAIL with ImportError for patch_language_prefs / LanguagePrefsIn.

• Step 3: Add the endpoint

In app/routers/api.py, near the existing patch_digest_prefs (around lines 868-897), add:

from app.services.i18n import ACTIVE_LANGUAGES


# ---------------------------------------------------------------------------
# Settings — language preference
# ---------------------------------------------------------------------------


class LanguagePrefsIn(BaseModel):
    lang: str


class LanguagePrefsOut(BaseModel):
    lang: str


@router.patch("/settings/language", response_model=LanguagePrefsOut)
async def patch_language_prefs(
    payload: LanguagePrefsIn,
    principal: CurrentUser = Depends(require_token),
    session: AsyncSession = Depends(get_session),
) -> LanguagePrefsOut:
    if principal.user is None:
        raise HTTPException(status_code=400, detail="no_user_context")
    lang = (payload.lang or "").strip().lower()
    if lang not in ACTIVE_LANGUAGES:
        raise HTTPException(
            status_code=400,
            detail=f"unsupported language: {payload.lang!r}",
        )
    user = await session.get(User, principal.user.id)
    if user is None:
        raise HTTPException(status_code=404, detail="user_not_found")
    user.lang = lang
    await session.commit()
    return LanguagePrefsOut(lang=lang)

(User and BaseModel are already imported at the top of app/routers/api.py. If ACTIVE_LANGUAGES import collides with anything else, alias it.)

• Step 4: Run tests to verify they pass

docker compose -f docker-compose.test.yml run --rm test pytest tests/test_localization_integration.py -v

Expected: all 12 tests pass.

• Step 5: Commit

git add app/routers/api.py tests/test_localization_integration.py
git commit -m "settings: PATCH /api/settings/language with ACTIVE_LANGUAGES gate"

---

Task 10: Settings UI dropdown

Files:

• Modify: app/templates/settings.html

• Modify: app/routers/pages.py::settings_page — pass user.lang to the template context

• Step 1: Add the language section to settings.html

In app/templates/settings.html, find the existing settings sections (<details class="settings-section"> blocks). Add a new section next to the email-digest preferences:

<details class="settings-section">
  <summary class="settings-section__head">Language</summary>
  <p class="settings-section__lede">
    Language the AI uses for the strategic log, your daily digest, and
    portfolio commentary. The interface itself stays in English for now.
  </p>
  <div class="settings-row">
    <select id="lang-select" class="settings-select">
      <option value="en" {% if (user.lang or 'en') == 'en' %}selected{% endif %}>English</option>
      <option value="it" {% if (user.lang or 'en') == 'it' %}selected{% endif %}>Italiano</option>
      <option value="es" disabled>Español &middot; coming soon</option>
      <option value="fr" disabled>Français &middot; coming soon</option>
      <option value="de" disabled>Deutsch &middot; coming soon</option>
    </select>
    <span id="lang-status" class="settings-status" aria-live="polite"></span>
  </div>
  <script>
    (function () {
      var sel = document.getElementById('lang-select');
      var status = document.getElementById('lang-status');
      if (!sel) return;
      sel.addEventListener('change', async function () {
        status.textContent = 'saving…';
        try {
          var r = await fetch('/api/settings/language', {
            method: 'PATCH',
            headers: {'Content-Type': 'application/json'},
            body: JSON.stringify({lang: sel.value}),
          });
          if (!r.ok) throw new Error('HTTP ' + r.status);
          status.textContent = '✓ saved';
          setTimeout(function () { status.textContent = ''; }, 1500);
        } catch (e) {
          status.textContent = '✗ failed';
        }
      });
    })();
  </script>
</details>

• Step 2: Confirm settings_page passes user.lang

In app/routers/pages.py::settings_page, the template context already includes user. The template reads user.lang directly from that object. No code change required — the Jinja2 expression {% if (user.lang or 'en') == 'it' %} handles old rows whose lang field hasn't been populated yet (defensive, post-migration).

If the existing template context does NOT pass the user object (it should, based on the digest-prefs section), add it.

• Step 3: Manual smoke verification step

Smoke is deferred to Task 11. No test step here — pure markup + small inline JS.

• Step 4: Commit

git add app/templates/settings.html
git commit -m "settings: add language dropdown (IT active, ES/FR/DE WIP)"

---

Task 11: Final regression + deploy + manual smoke

Files:

• (no code changes — verification only)

• Step 1: Full test suite

docker compose -f docker-compose.test.yml run --rm test pytest tests/ 2>&1 | tail -5

Expected: every previous test plus the new tests/test_i18n.py and tests/test_localization_integration.py pass. Total should now be ~280 passing.

• Step 2: Apply migration to prod DB (requires explicit user approval)

docker compose exec app alembic upgrade head

Expected: Running upgrade 0021 -> 0022, localization.

• Step 3: Restart prod app (requires explicit user approval)

docker compose restart app
docker compose logs app --tail 30 | grep -E "(Uvicorn|startup complete|ERROR|Traceback)"

Expected: Application startup complete. cleanly; no tracebacks.

• Step 4: Manual smoke — switch a paid test user to Italian

In a paid-tier browser session, open /settings. Confirm the Language dropdown appears with all five options, the English option selected, ES/FR/DE disabled and labelled "coming soon". Pick Italiano, confirm the inline ✓ saved status appears. Refresh — Italian remains selected.

• Step 5: Manual smoke — strategic log translation

Wait for the next hourly ai_log_job tick (or trigger via the scheduler/admin). Confirm a row appears in strategic_log_translations with lang='it'. NOTE: this requires a prod DB read; only run with explicit user approval:

docker compose exec app python -c "
import asyncio
from sqlalchemy import select
from app.db import get_session_factory
from app.models import StrategicLogTranslation

async def main():
    factory = get_session_factory()
    async with factory() as s:
        rows = (await s.execute(
            select(StrategicLogTranslation).order_by(StrategicLogTranslation.id.desc()).limit(3)
        )).scalars().all()
        for r in rows:
            print(r.id, r.log_id, r.lang, r.llm_model, r.llm_cost_usd, r.content_md[:80])
asyncio.run(main())
"

Expected: at least one row with lang='it', llm_model containing deepseek, llm_cost_usd a small positive number.

• Step 6: Manual smoke — portfolio analysis

On the dashboard as the Italian user, click "Analyse" (or whatever triggers /api/analyze). Confirm the rendered AI commentary is in Italian.

• Step 7: Manual smoke — email digest

docker compose exec app python -m app.cli send-test-digest <italian-user-email> daily

Expected: digest email lands in Italian, including the subject line.

• Step 8: Manual smoke — Edge cases

• Direct curl -X PATCH /api/settings/language with {"lang": "es"} → 400.

• Switch user back to English, refresh dashboard — log renders in English again.

---

Self-Review

Spec coverage walkthrough:

• users.lang column with default 'en' → Task 3 model + Task 4 migration
• strategic_log_translations table → Task 3 model + Task 4 migration
• i18n.LANGUAGES + ACTIVE_LANGUAGES → Task 1
• respond_in_clause() → Task 1
• translate() helper → Task 2 (no-op fast path for en/unknown; code-fence stripping; raises on provider failure)
• ai_log_job translation fan-out → Task 5 (parallel via asyncio.gather; per-language failure isolated)
• Portfolio analysis lang-aware system prompt → Task 6
• Email digest lang-aware per-user generation → Task 7
• /log localized fetch with English fallback → Task 8
• PATCH /api/settings/language with ACTIVE_LANGUAGES gate → Task 9
• Settings dropdown with IT active + ES/FR/DE disabled → Task 10
• No tier gating on translation → Task 5 query selects on User.lang only, no tier filter
• No retroactive backfill → not built; only forward-going translations
• No UI label translation → out of scope, Task 10 surfaces this in the section copy

Type / signature consistency:

• respond_in_clause(lang: str | None) -> str — used in Tasks 6, 7. Consistent.
• translate(client, text, target_lang) -> tuple[str, LogResult] — used in Tasks 2, 5. Consistent.
• ACTIVE_LANGUAGES: set[str] — used in Tasks 5, 9. Consistent.
• LanguagePrefsIn { lang: str } / LanguagePrefsOut { lang: str } — used in Task 9 only.
• _resolve_log_content(session, log_id, lang) -> str — used in Task 8 only.
• translate_log_for_active_languages(session, log_id) -> None — used in Task 5 only.

Note on Task 7: if _generate_variants is currently called ONCE for all users in the digest job (variants shared), the localization plan requires it to be called per-user. The plan flags this and asks the engineer to surface a concern rather than silently restructuring. If the structure differs from expectation, the engineer should escalate before proceeding.