Add generic agent activity tracking and Alembic migrations

CLAUDE.md

@@ -8,16 +8,16 @@ Guidance for working in this repository.
 
 ## Layout
 
-All application code lives under `src/`. Packages keep their top-level names (`agents`, `db`, `utils`), so imports are `from agents...` / `from db...` / `from utils...`, with `src` on the path (pytest sets `pythonpath = ["src"]`; the Docker image runs uvicorn with `--app-dir src`). Run entrypoints from the repo root so `load_dotenv()` finds `.env`.
+All application code lives under `src/`. Packages keep their top-level names (`agents`, `db`, `emails`), so imports are `from agents...` / `from db...` / `from emails...`, with `src` on the path (pytest sets `pythonpath = ["src"]`; the Docker image runs uvicorn with `--app-dir src`). Run entrypoints from the repo root so `load_dotenv()` finds `.env`.
 
 - `src/api.py` — FastAPI service. `POST /run` (full campaign) and `POST /test` (one user). Both run in the background, return `202`, and default to dry-run. Auth via the `X-API-Key` header matched against `SECRET_KEY`.
 - `src/app.py` — local CLI mirroring the API (`run`, `test`), for testing without HTTP.
 - `src/agents/orchestrator.py` — the pipeline: analyst → 5 parallel beat desks (researcher → writer → editor) → managing-editor gap roundtable → masthead → reviewer → deterministic clean/assemble/dedupe. Most non-agent logic (citation gating, URL/article validation, dedupe, subject-name canonicalization, prose humanizing) lives here.
-- `src/agents/` — one module per agent (`analyst`, `researcher`, `writer`, `editor`, `managing_editor`, `reviewer`), plus `beats.py` (beat desks), `campaign.py` (top-level run over subscriptions), `providers/` (subject-memory and ticker-profile context providers), and `tools/` (Serper search, web fetch).
+- `src/agents/` — one module per agent (`analyst`, `researcher`, `writer`, `editor`, `managing_editor`, `reviewer`), plus `beats.py` (beat desks), `campaign.py` (top-level run over subscriptions), `sections.py` (the five editorial beats), `providers/` (subject-memory and ticker-profile context providers), and `tools/` (Serper search, web fetch).
+- `src/agents/runtime/` — agent plumbing shared by every agent: `chat_client.py` (per-role chat client plus the `SKILLS` provider), `make_agent.py` (the factory that wires generic activity tracking into every agent), `guardrails.py` (guardrail/citation middleware), and `tracking.py` (the generic `ActivityTracker`/`ToolTracker` middleware, `newsletter_scope`, and run context vars). New agents are built via `make_agent(...)` so tracking is automatic.
 - `src/agents/skills/` — `SKILL.md` files that control agent behavior (`subject-profile`, `section-research`, `newsletter-format`). Prefer editing these over code when changing how agents research or write.
-- `src/db/` — all database access. `mediapulse.py` reads subscriptions and ticker profiles from the upstream MediaPulse Postgres (`MEDIAPULSE_DATABASE_URL`, read-only, raw psycopg). The app's own Postgres (`DATABASE_URL`, SQLModel) uses `engine.py` for the shared engine, with each table's model alongside its operations in `newsletters.py` (archives each generated newsletter as markdown plus JSONB metadata) and `memory.py` (subject-brief agent memory). Tables are auto-created on first write.
+- `src/db/` — all database access. `mediapulse.py` reads subscriptions and ticker profiles from the upstream MediaPulse Postgres (`MEDIAPULSE_DATABASE_URL`, read-only, raw psycopg). The app's own Postgres (`DATABASE_URL`, SQLModel) uses `engine.py` for the shared engine, with each table's model alongside its operations: `newsletters.py` (archives each newsletter as markdown plus JSONB metadata, with a `pending`/`complete`/`failed` lifecycle via `create_newsletter`/`finalize_newsletter`), `memory.py` (subject-brief agent memory), and `agent_activity.py` (one row per agent run and tool call, tied to its `newsletter_id`, recording status, duration, model, and token usage). The schema is owned by Alembic migrations, not `create_all` (see Migrations).
 - `src/emails/` — everything email-related. `mailer.py` sends via Resend. `templates/` pairs each template's renderer with its tokenized HTML: `templates/newsletter.py` parses the newsletter markdown and fills `templates/newsletter.html` (the gitignored build artifact from `email-playground`).
-- `src/utils/` — `client.py`, `guardrails.py`, `sections.py`.
 - `email-playground/` — a standalone React Email (TypeScript) project that is the visual source-of-truth for MediaPulse email templates (the newsletter is the first one). `npm run build:templates` renders each template to a tokenized `src/emails/templates/<name>.html` that the Python side consumes (`emails/templates/newsletter.py` for the newsletter). Those HTML files are gitignored build artifacts, regenerated fresh in CI and the Docker build. Restyle emails there, not in Python. See `email-playground/README.md`.
 - `tests/` — pytest suite covering the deterministic logic (see Testing below).
 
@@ -39,6 +39,19 @@ python src/app.py test --email=you@example.com      # dry-run one user
 python src/api.py                                    # serve at http://localhost:8000 (docs at /docs)
 ```
 
+## Migrations
+
+The app's own Postgres schema (`DATABASE_URL`) is managed by Alembic, not `create_all`. Migrations live in `alembic/versions/`; `alembic/env.py` puts `src` on the path, loads `.env`, resolves `DATABASE_URL` (reusing `db.engine._engine_url`), and targets `SQLModel.metadata`.
+
+```
+alembic upgrade head                                  # apply pending migrations
+alembic revision -m "add X"                            # new (hand-written) migration
+alembic revision --autogenerate -m "add X"            # diff models vs DB, needs a live DATABASE_URL
+alembic downgrade -1                                   # roll back one
+```
+
+The Docker image runs `alembic upgrade head` on container start via `docker-entrypoint.sh`, then execs the `CMD`. It is skipped when `DATABASE_URL` is unset or `RUN_MIGRATIONS=0`. To run migrations as a standalone release job with the same image: `docker run -e RUN_MIGRATIONS=0 <image> alembic upgrade head`. After changing a SQLModel table, add a migration in the same change so deploys stay in sync.
+
 ## Code quality
 
 CI (`.github/workflows/code-quality.yml`) runs exactly these two commands, both must pass:
@@ -67,7 +80,7 @@ cd email-playground && npm install && npm run build:templates
 
 Config is in `pyproject.toml` under `[tool.pytest.ini_options]`: `pythonpath = ["src"]`, `testpaths = ["tests"]`, `asyncio_mode = "auto"` (async tests need no decorator). CI runs `pytest` in a separate `tests` job after installing both requirement files.
 
-The suite covers the deterministic logic, not the LLM agents: orchestrator text/section helpers, the email template, Serper tools, the MediaPulse profile shaping, the newsletter store and subject-memory upsert, guardrail middleware, the subject-memory and ticker-profile context providers, the mailer, the client model resolution, and the campaign delivery flow. Conventions for new tests:
+The suite covers the deterministic logic, not the LLM agents: orchestrator text/section helpers, the email template, Serper tools, the MediaPulse profile shaping, the newsletter store and lifecycle and subject-memory upsert, the agent-activity store, the guardrail and activity-tracking middleware, the subject-memory and ticker-profile context providers, the mailer, the client model resolution, and the campaign delivery flow. Conventions for new tests:
 
 - `tests/conftest.py` sets dummy credentials before collection, because importing the agent modules constructs the whole agent graph at import time.
 - Every external call (httpx, psycopg, the LLM clients) is monkeypatched, and the SQLModel store is exercised against in-memory SQLite. The suite is fully offline. A test that reaches the network is a test bug.

Dockerfile

@@ -21,9 +21,12 @@ COPY . .
 COPY --from=template-builder /build/src/emails/templates/ src/emails/templates/
 
 RUN useradd --no-create-home --shell /bin/false app \
+    && chmod +x docker-entrypoint.sh \
     && chown -R app:app /app
 USER app
 
 EXPOSE 8000
 
+# Entrypoint applies migrations (alembic upgrade head) before launching the command.
+ENTRYPOINT ["./docker-entrypoint.sh"]
 CMD ["uvicorn", "api:app", "--app-dir", "src", "--host", "0.0.0.0", "--port", "8000"]

alembic.ini

@@ -0,0 +1,40 @@
+# Alembic configuration. The database URL is resolved from DATABASE_URL in alembic/env.py,
+# so sqlalchemy.url is intentionally left blank here.
+[alembic]
+script_location = alembic
+prepend_sys_path = .
+sqlalchemy.url =
+
+[loggers]
+keys = root,sqlalchemy,alembic
+
+[handlers]
+keys = console
+
+[formatters]
+keys = generic
+
+[logger_root]
+level = WARNING
+handlers = console
+qualname =
+
+[logger_sqlalchemy]
+level = WARNING
+handlers =
+qualname = sqlalchemy.engine
+
+[logger_alembic]
+level = INFO
+handlers =
+qualname = alembic
+
+[handler_console]
+class = StreamHandler
+args = (sys.stderr,)
+level = NOTSET
+formatter = generic
+
+[formatter_generic]
+format = %(levelname)-5.5s [%(name)s] %(message)s
+datefmt = %H:%M:%S

alembic/env.py

@@ -0,0 +1,64 @@
+import os
+import sys
+from logging.config import fileConfig
+from pathlib import Path
+
+from alembic import context
+from sqlalchemy import create_engine
+
+# Put `src` on the path so the app's db package imports the same way it does at runtime.
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from dotenv import load_dotenv
+from sqlmodel import SQLModel
+
+from db.agent_activity import AgentActivity
+from db.engine import _engine_url
+from db.memory import SubjectMemory
+from db.newsletters import Newsletter
+
+load_dotenv()
+
+# Register every model on SQLModel.metadata so autogenerate can see the full schema.
+_REGISTERED_MODELS = (Newsletter, SubjectMemory, AgentActivity)
+target_metadata = SQLModel.metadata
+
+config = context.config
+
+if config.config_file_name is not None:
+    fileConfig(config.config_file_name)
+
+
+def _database_url() -> str:
+    if not os.getenv("DATABASE_URL"):
+        raise RuntimeError("DATABASE_URL is not set; Alembic needs it to run migrations.")
+
+    return _engine_url()
+
+
+def run_migrations_offline() -> None:
+    context.configure(
+        url=_database_url(),
+        target_metadata=target_metadata,
+        literal_binds=True,
+        dialect_opts={"paramstyle": "named"},
+    )
+
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+def run_migrations_online() -> None:
+    engine = create_engine(_database_url(), pool_pre_ping=True)
+
+    with engine.connect() as connection:
+        context.configure(connection=connection, target_metadata=target_metadata)
+
+        with context.begin_transaction():
+            context.run_migrations()
+
+
+if context.is_offline_mode():
+    run_migrations_offline()
+else:
+    run_migrations_online()

alembic/script.py.mako

@@ -0,0 +1,27 @@
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision | comma,n}
+Create Date: ${create_date}
+
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+import sqlalchemy as sa
+${imports if imports else ""}
+
+# revision identifiers, used by Alembic.
+revision: str = ${repr(up_revision)}
+down_revision: str | None = ${repr(down_revision)}
+branch_labels: str | Sequence[str] | None = ${repr(branch_labels)}
+depends_on: str | Sequence[str] | None = ${repr(depends_on)}
+
+
+def upgrade() -> None:
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade() -> None:
+    ${downgrades if downgrades else "pass"}

alembic/versions/0001_initial.py

@@ -0,0 +1,74 @@
+"""initial schema: newsletters, subject_memory, agent_activity
+
+Revision ID: 0001_initial
+Revises:
+Create Date: 2026-06-18
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "0001_initial"
+down_revision: str | None = None
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "newsletters",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("subject", sa.String(), nullable=False),
+        sa.Column("content", sa.String(), nullable=False),
+        sa.Column("status", sa.String(), nullable=False),
+        sa.Column("metadata", postgresql.JSONB(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+    )
+    op.create_index("ix_newsletters_subject", "newsletters", ["subject"])
+    op.create_index("ix_newsletters_status", "newsletters", ["status"])
+
+    op.create_table(
+        "subject_memory",
+        sa.Column("subject", sa.String(), primary_key=True),
+        sa.Column("brief", sa.String(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+    )
+
+    op.create_table(
+        "agent_activity",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("newsletter_id", sa.Integer(), nullable=False),
+        sa.Column("kind", sa.String(), nullable=False),
+        sa.Column("name", sa.String(), nullable=False),
+        sa.Column("model", sa.String(), nullable=True),
+        sa.Column("status", sa.String(), nullable=False),
+        sa.Column("duration_ms", sa.Integer(), nullable=True),
+        sa.Column("input_tokens", sa.Integer(), nullable=True),
+        sa.Column("output_tokens", sa.Integer(), nullable=True),
+        sa.Column("total_tokens", sa.Integer(), nullable=True),
+        sa.Column("error", sa.String(), nullable=True),
+        sa.Column("metadata", postgresql.JSONB(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+    )
+    op.create_index("ix_agent_activity_newsletter_id", "agent_activity", ["newsletter_id"])
+    op.create_index("ix_agent_activity_name", "agent_activity", ["name"])
+    op.create_index("ix_agent_activity_model", "agent_activity", ["model"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_agent_activity_model", table_name="agent_activity")
+    op.drop_index("ix_agent_activity_name", table_name="agent_activity")
+    op.drop_index("ix_agent_activity_newsletter_id", table_name="agent_activity")
+    op.drop_table("agent_activity")
+
+    op.drop_table("subject_memory")
+
+    op.drop_index("ix_newsletters_status", table_name="newsletters")
+    op.drop_index("ix_newsletters_subject", table_name="newsletters")
+    op.drop_table("newsletters")

docker-entrypoint.sh

@@ -0,0 +1,11 @@
+#!/usr/bin/env sh
+set -e
+
+# Apply pending migrations on startup, unless disabled (RUN_MIGRATIONS=0) or no database is configured.
+# Set RUN_MIGRATIONS=0 and pass `alembic upgrade head` as the command to run migrations as a standalone job.
+if [ "${RUN_MIGRATIONS:-1}" != "0" ] && [ -n "$DATABASE_URL" ]; then
+    echo "Applying database migrations (alembic upgrade head)..."
+    alembic upgrade head
+fi
+
+exec "$@"

environment.yml

@@ -10,5 +10,6 @@ dependencies:
       - httpx
       - psycopg[binary]
       - sqlmodel
+      - alembic
       - fastapi
       - uvicorn

pyproject.toml

@@ -8,6 +8,7 @@ select = ["E", "F", "I", "W"]
 [tool.ruff.lint.per-file-ignores]
 "src/api.py" = ["E402"]
 "src/app.py" = ["E402"]
+"alembic/env.py" = ["E402", "I001"]
 
 [tool.pytest.ini_options]
 pythonpath = ["src"]

requirements.txt

@@ -3,5 +3,6 @@ python-dotenv
 httpx
 psycopg[binary]
 sqlmodel
+alembic
 fastapi
 uvicorn

src/agents/analyst.py

@@ -1,12 +1,11 @@
-from agent_framework import Agent
-
 from agents.providers.memory import SubjectMemoryProvider
 from agents.providers.ticker import TickerProfileProvider
+from agents.runtime.chat_client import SKILLS, chat_client
+from agents.runtime.guardrails import SubjectGuardrail
+from agents.runtime.make_agent import make_agent
 from agents.tools import search, web_fetch
-from utils.client import SKILLS, chat_client
-from utils.guardrails import SubjectGuardrail
 
-analyst = Agent(
+analyst = make_agent(
     name="analyst",
     description="Resolves a subject into a research brief.",
     client=chat_client("analyst"),

src/agents/beats.py

@@ -3,9 +3,9 @@
 from agent_framework import Agent
 
 from agents.researcher import make_researcher
+from agents.runtime.guardrails import SourceRegistry
+from agents.sections import SECTIONS, Section
 from agents.writer import make_writer
-from utils.guardrails import SourceRegistry
-from utils.sections import SECTIONS, Section
 
 
 @dataclass

src/agents/campaign.py

@@ -2,8 +2,9 @@
 import time
 
 from agents.orchestrator import run_newsletter
+from agents.runtime.tracking import newsletter_scope
 from db.mediapulse import fetch_subscriptions
-from db.newsletters import save_newsletter
+from db.newsletters import create_newsletter, finalize_newsletter
 from emails.mailer import send_email
 from emails.templates.newsletter import newsletter_sources, render_newsletter_email
 
@@ -29,10 +30,12 @@ async def run_campaign(*, subscriptions: list[dict] | None = None, send: bool =
     last_send = 0.0
     delivered: list[dict] = []
 
-    async def deliver(ticker: str, recipients: list[dict], markdown: str) -> None:
+    async def deliver(ticker: str, recipients: list[dict], markdown: str, newsletter_id: int | None) -> None:
         nonlocal last_send
         email = render_newsletter_email(markdown, ticker=ticker)
-        save_newsletter(ticker, markdown, {"ticker": ticker, "sources": newsletter_sources(markdown)})
+        finalize_newsletter(
+            newsletter_id, content=markdown, metadata={"ticker": ticker, "sources": newsletter_sources(markdown)}
+        )
 
         for subscription in recipients:
             if send:
@@ -53,19 +56,23 @@ async def deliver(ticker: str, recipients: list[dict], markdown: str) -> None:
             delivered.append({"email": subscription["email"], "ticker": ticker})
 
     async def process(ticker: str, recipients: list[dict]) -> None:
-        async with generate_semaphore:
-            log(f"generating {ticker} ...")
+        newsletter_id = create_newsletter(ticker)
 
-            try:
-                markdown = await run_newsletter(ticker)
-            except Exception as error:
-                log(f"failed {ticker}: {error}")
+        with newsletter_scope(newsletter_id):
+            async with generate_semaphore:
+                log(f"generating {ticker} ...")
 
-                return
+                try:
+                    markdown = await run_newsletter(ticker)
+                except Exception as error:
+                    finalize_newsletter(newsletter_id, status="failed")
+                    log(f"failed {ticker}: {error}")
 
-            log(f"done {ticker}")
+                    return
 
-        await deliver(ticker, recipients, markdown)
+                log(f"done {ticker}")
+
+            await deliver(ticker, recipients, markdown, newsletter_id)
 
     await asyncio.gather(*(process(ticker, recipients) for ticker, recipients in subscribers_by_ticker.items()))