firefly/docs/superpowers/plans/2026-08-23-phase1-foundation.md

Phase 1: Foundation — 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: Stand up the agent worker container — config, agentdb with migrations, a read-only Firefly API client, a scheduled heartbeat that proves end-to-end connectivity — and deploy it to the production host.

Architecture: A Python 3.12 package under agent/ in this repo, run as one container in the existing compose stack. On startup it creates/migrates agentdb (a second database in the existing Postgres container), runs a heartbeat once, then hands off to APScheduler. The heartbeat calls Firefly's /api/v1/about and records the outcome in sync_runs — the same logging path every future connector will use.

Tech Stack: Python 3.12, APScheduler 3.x, httpx, psycopg 3, pytest, testcontainers (dev only), Docker/compose.

Spec: docs/superpowers/specs/2026-08-23-accounting-agent-architecture-design.md

Global Constraints

• Python 3.12+ (requires-python = ">=3.12"); container base image python:3.12-slim.
• Runtime dependencies limited to: apscheduler>=3.10,<4, httpx>=0.27, psycopg[binary]>=3.1. Dev-only: pytest>=8, testcontainers[postgres]>=4.
• TDD: every code task writes the failing test first, sees it fail, then implements.
• Firefly API access in this phase is read-only (GET endpoints only). The agent never writes to the ledger.
• Secrets live only in .env on the host. Never commit .env, tokens, or passwords.
• All timestamps are UTC (TIMESTAMPTZ in Postgres, datetime.now(timezone.utc) in Python).
• The agent container attaches only to the internal firefly network: no Traefik labels, no published ports.
• Commit after every green test cycle.
• Prerequisite for running tests locally: Docker daemon running (testcontainers starts a throwaway postgres:16-alpine).
• Testing deviation from spec, by design: the spec's throwaway-Firefly compose test profile arrives in Phase 2 with the sync engine. Phase 1's client makes two GET calls, tested against httpx.MockTransport; Task 9's agent.verify exercises the real API in production.

File Structure

agent/
├── Dockerfile
├── .dockerignore
├── pyproject.toml
├── README.md                     # setup + deploy runbook (Task 8)
├── src/agent/
│   ├── __init__.py               # __version__
│   ├── __main__.py               # entrypoint: ensure db → migrate → heartbeat → scheduler
│   ├── config.py                 # AgentConfig, load_config, ConfigError
│   ├── db.py                     # DSNs, ensure_database, run_migrations, connect
│   ├── migrations/0001_initial.sql
│   ├── firefly.py                # FireflyClient (read-only), FireflyError
│   ├── scheduler.py              # build_scheduler
│   ├── healthcheck.py            # Docker HEALTHCHECK entry
│   ├── verify.py                 # `python -m agent.verify` post-deploy check
│   └── jobs/
│       ├── __init__.py
│       └── heartbeat.py          # run_heartbeat
└── tests/
    ├── conftest.py               # session postgres container, per-test cfg
    ├── test_smoke.py
    ├── test_config.py
    ├── test_db.py
    ├── test_firefly.py
    ├── test_heartbeat.py
    ├── test_scheduler.py
    └── test_verify.py

Everything runs from agent/ (cd agent && pytest). Repo-root files touched: docker-compose.yml, .env.example, .gitignore.

---

Task 1: Package scaffold

Files:

• Create: agent/pyproject.toml, agent/src/agent/__init__.py, agent/src/agent/jobs/__init__.py, agent/tests/test_smoke.py
• Modify: .gitignore

Interfaces:

• Produces: importable package agent with agent.__version__ == "0.1.0"; working pytest invocation from agent/.

• Step 1: Create agent/pyproject.toml

[project]
name = "firefly-agent"
version = "0.1.0"
description = "Personal accounting agent worker for the Firefly III stack"
requires-python = ">=3.12"
dependencies = [
    "apscheduler>=3.10,<4",
    "httpx>=0.27",
    "psycopg[binary]>=3.1",
]

[project.optional-dependencies]
dev = [
    "pytest>=8",
    "testcontainers[postgres]>=4",
]

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

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

[tool.pytest.ini_options]
testpaths = ["tests"]

• Step 2: Write the failing smoke test — agent/tests/test_smoke.py

def test_package_imports():
    import agent

    assert agent.__version__ == "0.1.0"

• Step 3: Set up the venv and run the test to verify it fails

cd /home/jake/dev/firefly
python3 -m venv .venv
source .venv/bin/activate
pip install -e './agent[dev]'
cd agent && pytest tests/test_smoke.py -v

Expected: FAIL (agent has no attribute __version__ — the editable install will fail or the import will, since src/agent/__init__.py doesn't exist yet; create empty dirs as needed to get the install through, then see the assertion fail).

• Step 4: Implement — agent/src/agent/__init__.py

__version__ = "0.1.0"

Also create empty agent/src/agent/jobs/__init__.py.

• Step 5: Run test to verify it passes

Run: cd agent && pytest tests/test_smoke.py -v — Expected: PASS

• Step 6: Add ignores to repo-root .gitignore

Append these lines:

.venv/
__pycache__/
*.egg-info/
.pytest_cache/

• Step 7: Commit

git add agent/ .gitignore
git commit -m "feat(agent): scaffold Python package for the accounting agent worker"

---

Task 2: Config loading

Files:

• Create: agent/src/agent/config.py
• Test: agent/tests/test_config.py

Interfaces:

• Produces:
  • class ConfigError(ValueError)
  • @dataclass(frozen=True) AgentConfig with fields: db_host: str, db_port: int, db_user: str, db_password: str, agent_db_name: str, firefly_url: str, firefly_token: str, heartbeat_interval_minutes: int
  • load_config(env: Mapping[str, str] | None = None) -> AgentConfig — reads os.environ when env is None.

Environment variables (shared names match the existing .env): required — DB_HOST, POSTGRES_USER, POSTGRES_PASSWORD, FIREFLY_API_URL, AGENT_FIREFLY_TOKEN; optional with defaults — DB_PORT=5432, AGENT_DB_NAME=agentdb, HEARTBEAT_INTERVAL_MINUTES=60.

• Step 1: Write the failing tests — agent/tests/test_config.py

import pytest

from agent.config import AgentConfig, ConfigError, load_config

FULL_ENV = {
    "DB_HOST": "db",
    "DB_PORT": "5433",
    "POSTGRES_USER": "firefly",
    "POSTGRES_PASSWORD": "s3cret",
    "AGENT_DB_NAME": "agentdb2",
    "FIREFLY_API_URL": "http://app:8080",
    "AGENT_FIREFLY_TOKEN": "tok123",
    "HEARTBEAT_INTERVAL_MINUTES": "15",
}

MINIMAL_ENV = {
    "DB_HOST": "db",
    "POSTGRES_USER": "firefly",
    "POSTGRES_PASSWORD": "s3cret",
    "FIREFLY_API_URL": "http://app:8080",
    "AGENT_FIREFLY_TOKEN": "tok123",
}


def test_load_config_reads_all_vars():
    cfg = load_config(FULL_ENV)
    assert cfg == AgentConfig(
        db_host="db",
        db_port=5433,
        db_user="firefly",
        db_password="s3cret",
        agent_db_name="agentdb2",
        firefly_url="http://app:8080",
        firefly_token="tok123",
        heartbeat_interval_minutes=15,
    )


def test_defaults_applied():
    cfg = load_config(MINIMAL_ENV)
    assert cfg.db_port == 5432
    assert cfg.agent_db_name == "agentdb"
    assert cfg.heartbeat_interval_minutes == 60


@pytest.mark.parametrize(
    "missing",
    ["DB_HOST", "POSTGRES_USER", "POSTGRES_PASSWORD", "FIREFLY_API_URL", "AGENT_FIREFLY_TOKEN"],
)
def test_missing_required_var_names_the_var(missing):
    env = {k: v for k, v in MINIMAL_ENV.items() if k != missing}
    with pytest.raises(ConfigError, match=missing):
        load_config(env)

• Step 2: Run tests to verify they fail

Run: cd agent && pytest tests/test_config.py -v — Expected: FAIL with ModuleNotFoundError: No module named 'agent.config'

• Step 3: Implement — agent/src/agent/config.py

"""Agent configuration loaded from environment variables."""

from __future__ import annotations

import os
from collections.abc import Mapping
from dataclasses import dataclass


class ConfigError(ValueError):
    """A required environment variable is missing or invalid."""


@dataclass(frozen=True)
class AgentConfig:
    db_host: str
    db_port: int
    db_user: str
    db_password: str
    agent_db_name: str
    firefly_url: str
    firefly_token: str
    heartbeat_interval_minutes: int


def _require(env: Mapping[str, str], name: str) -> str:
    value = env.get(name, "").strip()
    if not value:
        raise ConfigError(f"missing required environment variable: {name}")
    return value


def load_config(env: Mapping[str, str] | None = None) -> AgentConfig:
    if env is None:
        env = os.environ
    return AgentConfig(
        db_host=_require(env, "DB_HOST"),
        db_port=int(env.get("DB_PORT", "5432")),
        db_user=_require(env, "POSTGRES_USER"),
        db_password=_require(env, "POSTGRES_PASSWORD"),
        agent_db_name=env.get("AGENT_DB_NAME", "agentdb"),
        firefly_url=_require(env, "FIREFLY_API_URL"),
        firefly_token=_require(env, "AGENT_FIREFLY_TOKEN"),
        heartbeat_interval_minutes=int(env.get("HEARTBEAT_INTERVAL_MINUTES", "60")),
    )

• Step 4: Run tests to verify they pass

Run: cd agent && pytest tests/test_config.py -v — Expected: PASS (7 tests)

• Step 5: Commit

git add agent/src/agent/config.py agent/tests/test_config.py
git commit -m "feat(agent): load config from environment with validation"

---

Task 3: agentdb — database bootstrap and migrations

Files:

• Create: agent/src/agent/db.py, agent/src/agent/migrations/0001_initial.sql, agent/tests/conftest.py
• Test: agent/tests/test_db.py

Interfaces:

• Consumes: AgentConfig from Task 2.

• Produces:

  • admin_dsn(cfg: AgentConfig) -> str — DSN to the always-present postgres maintenance database.
  • agent_dsn(cfg: AgentConfig) -> str — DSN to cfg.agent_db_name.
  • ensure_database(cfg: AgentConfig) -> None — creates agentdb if absent; idempotent.
  • run_migrations(cfg: AgentConfig) -> list[str] — applies pending migrations/*.sql in name order, returns the names applied (empty list when up to date).
  • connect(cfg: AgentConfig) -> psycopg.Connection — connection to agentdb.
  • Tables after migration 0001: schema_migrations, sync_runs, documents, bills, approvals.

• Test fixtures produced for later tasks (in conftest.py): pg (session-scoped Postgres container), cfg (per-test AgentConfig pointing at a unique database name in that container).

• Step 1: Write the migration SQL — agent/src/agent/migrations/0001_initial.sql

This is data, not logic — write it before the tests that assert on it. Schema is the spec's "agentdb schema (v1)" verbatim:

-- agentdb schema v1 (see spec: agentdb schema section)

CREATE TABLE sync_runs (
    id             BIGSERIAL PRIMARY KEY,
    connector      TEXT        NOT NULL,
    started_at     TIMESTAMPTZ NOT NULL,
    finished_at    TIMESTAMPTZ,
    status         TEXT        NOT NULL CHECK (status IN ('running', 'ok', 'error')),
    items_upserted INTEGER     NOT NULL DEFAULT 0,
    detail         TEXT
);

CREATE INDEX sync_runs_connector_id_idx ON sync_runs (connector, id DESC);

CREATE TABLE documents (
    id          BIGSERIAL PRIMARY KEY,
    path        TEXT        NOT NULL UNIQUE,
    sha256      CHAR(64)    NOT NULL UNIQUE,
    institution TEXT        NOT NULL,
    account     TEXT,
    doc_type    TEXT        NOT NULL,
    period      TEXT,
    source      TEXT        NOT NULL,
    received_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE bills (
    id              BIGSERIAL PRIMARY KEY,
    payee           TEXT          NOT NULL,
    expected_amount NUMERIC(12,2),
    due_date        DATE          NOT NULL,
    funding_account TEXT,
    autopay         BOOLEAN       NOT NULL DEFAULT false,
    status          TEXT          NOT NULL DEFAULT 'upcoming'
                    CHECK (status IN ('upcoming', 'approved', 'paid', 'skipped', 'overdue'))
);

CREATE TABLE approvals (
    id                  BIGSERIAL PRIMARY KEY,
    bill_id             BIGINT        NOT NULL REFERENCES bills (id),
    amount              NUMERIC(12,2) NOT NULL,
    requested_at        TIMESTAMPTZ   NOT NULL DEFAULT now(),
    decided_at          TIMESTAMPTZ,
    decision            TEXT          CHECK (decision IN ('approved', 'denied')),
    telegram_message_id TEXT,
    executed_at         TIMESTAMPTZ
);

(schema_migrations is created by the runner itself, not by a migration.)

• Step 2: Write the shared test fixtures — agent/tests/conftest.py

import itertools

import pytest
from testcontainers.postgres import PostgresContainer

from agent.config import AgentConfig

_db_counter = itertools.count()


@pytest.fixture(scope="session")
def pg():
    with PostgresContainer("postgres:16-alpine") as container:
        yield container


@pytest.fixture
def cfg(pg) -> AgentConfig:
    """AgentConfig pointing at a fresh, uniquely named database per test."""
    return AgentConfig(
        db_host=pg.get_container_host_ip(),
        db_port=int(pg.get_exposed_port(5432)),
        db_user=pg.username,
        db_password=pg.password,
        agent_db_name=f"agentdb_test_{next(_db_counter)}",
        firefly_url="http://firefly.test",
        firefly_token="test-token",
        heartbeat_interval_minutes=60,
    )

• Step 3: Write the failing tests — agent/tests/test_db.py

from agent.db import connect, ensure_database, run_migrations

EXPECTED_TABLES = {"schema_migrations", "sync_runs", "documents", "bills", "approvals"}


def test_ensure_database_creates_and_is_idempotent(cfg):
    ensure_database(cfg)
    ensure_database(cfg)  # second call must not raise
    with connect(cfg) as conn:
        assert conn.execute("SELECT 1").fetchone()[0] == 1


def test_run_migrations_applies_each_migration_exactly_once(cfg):
    ensure_database(cfg)

    first = run_migrations(cfg)
    assert first == ["0001_initial.sql"]

    second = run_migrations(cfg)
    assert second == []

    with connect(cfg) as conn:
        rows = conn.execute(
            "SELECT tablename FROM pg_tables WHERE schemaname = 'public'"
        ).fetchall()
    assert EXPECTED_TABLES <= {r[0] for r in rows}


def test_sync_runs_rejects_bad_status(cfg):
    import psycopg
    import pytest

    ensure_database(cfg)
    run_migrations(cfg)
    with connect(cfg) as conn, pytest.raises(psycopg.errors.CheckViolation):
        conn.execute(
            "INSERT INTO sync_runs (connector, started_at, status)"
            " VALUES ('x', now(), 'bogus')"
        )

• Step 4: Run tests to verify they fail

Run: cd agent && pytest tests/test_db.py -v — Expected: FAIL with ModuleNotFoundError: No module named 'agent.db' (the container may take ~10s to start first run)

• Step 5: Implement — agent/src/agent/db.py

"""agentdb access: bootstrap, migrations, connections."""

from __future__ import annotations

from pathlib import Path

import psycopg

from agent.config import AgentConfig

_MIGRATIONS_DIR = Path(__file__).parent / "migrations"


def _dsn(cfg: AgentConfig, dbname: str) -> str:
    return (
        f"host={cfg.db_host} port={cfg.db_port} dbname={dbname} "
        f"user={cfg.db_user} password={cfg.db_password}"
    )


def admin_dsn(cfg: AgentConfig) -> str:
    """DSN for the maintenance database that always exists."""
    return _dsn(cfg, "postgres")


def agent_dsn(cfg: AgentConfig) -> str:
    return _dsn(cfg, cfg.agent_db_name)


def connect(cfg: AgentConfig) -> psycopg.Connection:
    return psycopg.connect(agent_dsn(cfg))


def ensure_database(cfg: AgentConfig) -> None:
    """Create the agent database if it does not exist. Idempotent."""
    with psycopg.connect(admin_dsn(cfg), autocommit=True) as conn:
        exists = conn.execute(
            "SELECT 1 FROM pg_database WHERE datname = %s", (cfg.agent_db_name,)
        ).fetchone()
        if not exists:
            # CREATE DATABASE cannot be parameterized; the name comes from
            # our own config, quoted defensively via psycopg's identifier API.
            from psycopg import sql

            conn.execute(
                sql.SQL("CREATE DATABASE {}").format(sql.Identifier(cfg.agent_db_name))
            )


def run_migrations(cfg: AgentConfig) -> list[str]:
    """Apply pending migrations in filename order. Returns names applied."""
    applied: list[str] = []
    with connect(cfg) as conn:
        conn.execute(
            "CREATE TABLE IF NOT EXISTS schema_migrations ("
            " name TEXT PRIMARY KEY,"
            " applied_at TIMESTAMPTZ NOT NULL DEFAULT now())"
        )
        done = {
            row[0] for row in conn.execute("SELECT name FROM schema_migrations")
        }
        for path in sorted(_MIGRATIONS_DIR.glob("*.sql")):
            if path.name in done:
                continue
            conn.execute(path.read_text())
            conn.execute(
                "INSERT INTO schema_migrations (name) VALUES (%s)", (path.name,)
            )
            applied.append(path.name)
    return applied

• Step 6: Run tests to verify they pass

Run: cd agent && pytest tests/test_db.py -v — Expected: PASS (3 tests)

• Step 7: Commit

git add agent/src/agent/db.py agent/src/agent/migrations/ agent/tests/conftest.py agent/tests/test_db.py
git commit -m "feat(agent): agentdb bootstrap, migration runner, v1 schema"

---

Task 4: Firefly API client (read-only)

Files:

• Create: agent/src/agent/firefly.py
• Test: agent/tests/test_firefly.py

Interfaces:

• Produces:

  • class FireflyError(Exception)
  • class FireflyClient with:
    • __init__(self, base_url: str, token: str, transport: httpx.BaseTransport | None = None) — transport exists for tests only.
    • about(self) -> dict — returns the data object from GET /api/v1/about (contains version).
    • accounts(self, type: str = "asset") -> list[dict] — all pages of GET /api/v1/accounts; each item is Firefly's account object ({"id": ..., "attributes": {"name": ...}}).

• Step 1: Write the failing tests — agent/tests/test_firefly.py

import httpx
import pytest

from agent.firefly import FireflyClient, FireflyError


def make_client(handler) -> FireflyClient:
    return FireflyClient(
        "http://firefly.test", "tok", transport=httpx.MockTransport(handler)
    )


def test_about_sends_bearer_token_and_returns_data():
    def handler(request: httpx.Request) -> httpx.Response:
        assert request.headers["Authorization"] == "Bearer tok"
        assert request.headers["Accept"] == "application/json"
        assert request.url.path == "/api/v1/about"
        return httpx.Response(200, json={"data": {"version": "6.1.1"}})

    assert make_client(handler).about() == {"version": "6.1.1"}


def test_non_200_raises_firefly_error_with_status():
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(401, json={"message": "Unauthenticated"})

    with pytest.raises(FireflyError, match="401"):
        make_client(handler).about()


def test_accounts_walks_all_pages():
    def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path == "/api/v1/accounts"
        assert request.url.params["type"] == "asset"
        page = int(request.url.params.get("page", "1"))
        data = {
            1: [{"id": "1", "attributes": {"name": "NFCU Checking"}}],
            2: [{"id": "2", "attributes": {"name": "Venmo"}}],
        }[page]
        return httpx.Response(
            200,
            json={
                "data": data,
                "meta": {"pagination": {"total_pages": 2, "current_page": page}},
            },
        )

    accounts = make_client(handler).accounts()
    assert [a["attributes"]["name"] for a in accounts] == ["NFCU Checking", "Venmo"]

• Step 2: Run tests to verify they fail

Run: cd agent && pytest tests/test_firefly.py -v — Expected: FAIL with ModuleNotFoundError: No module named 'agent.firefly'

• Step 3: Implement — agent/src/agent/firefly.py

"""Minimal read-only client for the Firefly III REST API."""

from __future__ import annotations

import httpx


class FireflyError(Exception):
    """Firefly III returned an unexpected response."""


class FireflyClient:
    def __init__(
        self,
        base_url: str,
        token: str,
        transport: httpx.BaseTransport | None = None,
    ) -> None:
        self._http = httpx.Client(
            base_url=base_url,
            headers={
                "Authorization": f"Bearer {token}",
                "Accept": "application/json",
            },
            timeout=30.0,
            transport=transport,
        )

    def _get(self, path: str, params: dict | None = None) -> dict:
        response = self._http.get(path, params=params)
        if response.status_code != 200:
            raise FireflyError(
                f"GET {path} -> {response.status_code}: {response.text[:200]}"
            )
        return response.json()

    def about(self) -> dict:
        return self._get("/api/v1/about")["data"]

    def accounts(self, type: str = "asset") -> list[dict]:
        results: list[dict] = []
        page = 1
        while True:
            body = self._get("/api/v1/accounts", params={"type": type, "page": page})
            results.extend(body["data"])
            pagination = body.get("meta", {}).get("pagination", {})
            if page >= pagination.get("total_pages", 1):
                return results
            page += 1

• Step 4: Run tests to verify they pass

Run: cd agent && pytest tests/test_firefly.py -v — Expected: PASS (3 tests)

• Step 5: Commit

git add agent/src/agent/firefly.py agent/tests/test_firefly.py
git commit -m "feat(agent): read-only Firefly III API client"

---

Task 5: Heartbeat job

Files:

• Create: agent/src/agent/jobs/heartbeat.py
• Test: agent/tests/test_heartbeat.py

Interfaces:

• Consumes: AgentConfig (Task 2), connect (Task 3), FireflyClient/FireflyError (Task 4).

• Produces: run_heartbeat(cfg: AgentConfig, client: FireflyClient | None = None) -> str — calls Firefly about(), inserts one sync_runs row (connector='heartbeat', status 'ok' or 'error'), returns the status string. Builds its own client from cfg when client is None; the parameter exists for tests.

• Step 1: Write the failing tests — agent/tests/test_heartbeat.py

from agent.db import connect, ensure_database, run_migrations
from agent.firefly import FireflyError
from agent.jobs.heartbeat import run_heartbeat


class FakeFirefly:
    def __init__(self, version: str | None = None, error: Exception | None = None):
        self._version = version
        self._error = error

    def about(self) -> dict:
        if self._error:
            raise self._error
        return {"version": self._version}


def _last_run(cfg):
    with connect(cfg) as conn:
        return conn.execute(
            "SELECT connector, status, detail, started_at, finished_at"
            " FROM sync_runs ORDER BY id DESC LIMIT 1"
        ).fetchone()


def test_heartbeat_records_ok(cfg):
    ensure_database(cfg)
    run_migrations(cfg)

    status = run_heartbeat(cfg, client=FakeFirefly(version="6.1.1"))

    assert status == "ok"
    connector, db_status, detail, started_at, finished_at = _last_run(cfg)
    assert connector == "heartbeat"
    assert db_status == "ok"
    assert "6.1.1" in detail
    assert finished_at >= started_at


def test_heartbeat_records_error_and_does_not_raise(cfg):
    ensure_database(cfg)
    run_migrations(cfg)

    status = run_heartbeat(cfg, client=FakeFirefly(error=FireflyError("GET /api/v1/about -> 401")))

    assert status == "error"
    connector, db_status, detail, *_ = _last_run(cfg)
    assert db_status == "error"
    assert "401" in detail

• Step 2: Run tests to verify they fail

Run: cd agent && pytest tests/test_heartbeat.py -v — Expected: FAIL with ModuleNotFoundError: No module named 'agent.jobs.heartbeat'

• Step 3: Implement — agent/src/agent/jobs/heartbeat.py

"""Heartbeat: prove Firefly API + agentdb connectivity on a schedule."""

from __future__ import annotations

import logging
from datetime import datetime, timezone

from agent.config import AgentConfig
from agent.db import connect
from agent.firefly import FireflyClient

logger = logging.getLogger(__name__)


def run_heartbeat(cfg: AgentConfig, client: FireflyClient | None = None) -> str:
    if client is None:
        client = FireflyClient(cfg.firefly_url, cfg.firefly_token)

    started_at = datetime.now(timezone.utc)
    try:
        info = client.about()
        status, detail = "ok", f"firefly version {info.get('version', 'unknown')}"
    except Exception as exc:  # any failure is a recorded outcome, not a crash
        status, detail = "error", f"{type(exc).__name__}: {exc}"
        logger.warning("heartbeat failed: %s", detail)

    finished_at = datetime.now(timezone.utc)
    with connect(cfg) as conn:
        conn.execute(
            "INSERT INTO sync_runs (connector, started_at, finished_at, status, detail)"
            " VALUES (%s, %s, %s, %s, %s)",
            ("heartbeat", started_at, finished_at, status, detail),
        )
    return status

• Step 4: Run tests to verify they pass

Run: cd agent && pytest tests/test_heartbeat.py -v — Expected: PASS (2 tests)

• Step 5: Commit

git add agent/src/agent/jobs/heartbeat.py agent/tests/test_heartbeat.py
git commit -m "feat(agent): heartbeat job recording to sync_runs"

---

Task 6: Scheduler, healthcheck, entrypoint

Files:

• Create: agent/src/agent/scheduler.py, agent/src/agent/healthcheck.py, agent/src/agent/__main__.py
• Test: agent/tests/test_scheduler.py

Interfaces:

• Consumes: AgentConfig, ensure_database, run_migrations, connect, run_heartbeat.

• Produces:

  • build_scheduler(cfg: AgentConfig, *, scheduler_class: type = BlockingScheduler) -> BaseScheduler — registers job id "heartbeat" on an interval of cfg.heartbeat_interval_minutes minutes. scheduler_class exists for tests (BackgroundScheduler).
  • python -m agent — startup sequence: load config → ensure_database → run_migrations → run_heartbeat once → scheduler runs forever.
  • python -m agent.healthcheck — exit 0 when agentdb answers SELECT 1, exit 1 otherwise (Docker HEALTHCHECK).

• Step 1: Write the failing test — agent/tests/test_scheduler.py

from datetime import timedelta

from apscheduler.schedulers.background import BackgroundScheduler

from agent.config import AgentConfig
from agent.scheduler import build_scheduler


def _cfg(minutes: int) -> AgentConfig:
    return AgentConfig(
        db_host="db",
        db_port=5432,
        db_user="u",
        db_password="p",
        agent_db_name="agentdb",
        firefly_url="http://app:8080",
        firefly_token="tok",
        heartbeat_interval_minutes=minutes,
    )


def test_build_scheduler_registers_heartbeat_at_configured_interval():
    scheduler = build_scheduler(_cfg(minutes=15), scheduler_class=BackgroundScheduler)
    scheduler.start(paused=True)
    try:
        job = scheduler.get_job("heartbeat")
        assert job is not None
        assert job.trigger.interval == timedelta(minutes=15)
    finally:
        scheduler.shutdown(wait=False)

• Step 2: Run test to verify it fails

Run: cd agent && pytest tests/test_scheduler.py -v — Expected: FAIL with ModuleNotFoundError: No module named 'agent.scheduler'

• Step 3: Implement — agent/src/agent/scheduler.py

"""APScheduler assembly: one place where jobs get registered."""

from __future__ import annotations

from apscheduler.schedulers.base import BaseScheduler
from apscheduler.schedulers.blocking import BlockingScheduler

from agent.config import AgentConfig
from agent.jobs.heartbeat import run_heartbeat


def build_scheduler(
    cfg: AgentConfig, *, scheduler_class: type = BlockingScheduler
) -> BaseScheduler:
    scheduler = scheduler_class(timezone="UTC")
    scheduler.add_job(
        run_heartbeat,
        "interval",
        args=[cfg],
        minutes=cfg.heartbeat_interval_minutes,
        id="heartbeat",
    )
    return scheduler

• Step 4: Run test to verify it passes

Run: cd agent && pytest tests/test_scheduler.py -v — Expected: PASS

• Step 5: Implement the healthcheck — agent/src/agent/healthcheck.py

Glue over tested parts; no dedicated test.

"""Docker HEALTHCHECK entry: healthy = agentdb answers SELECT 1."""

from __future__ import annotations

import sys

from agent.config import load_config
from agent.db import connect


def main() -> int:
    try:
        cfg = load_config()
        with connect(cfg) as conn:
            conn.execute("SELECT 1")
        return 0
    except Exception as exc:
        print(f"unhealthy: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    sys.exit(main())

• Step 6: Implement the entrypoint — agent/src/agent/__main__.py

"""Agent entrypoint: bootstrap the database, then run scheduled jobs forever."""

from __future__ import annotations

import logging

from agent import __version__
from agent.config import load_config
from agent.db import ensure_database, run_migrations
from agent.jobs.heartbeat import run_heartbeat
from agent.scheduler import build_scheduler

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
logger = logging.getLogger("agent")


def main() -> None:
    logger.info("firefly-agent %s starting", __version__)
    cfg = load_config()
    ensure_database(cfg)
    applied = run_migrations(cfg)
    if applied:
        logger.info("applied migrations: %s", ", ".join(applied))
    status = run_heartbeat(cfg)
    logger.info("startup heartbeat: %s", status)
    logger.info(
        "scheduler starting; heartbeat every %d minutes",
        cfg.heartbeat_interval_minutes,
    )
    build_scheduler(cfg).start()


if __name__ == "__main__":
    main()

• Step 7: Run the full suite

Run: cd agent && pytest -v — Expected: all tests PASS

• Step 8: Commit

git add agent/src/agent/scheduler.py agent/src/agent/healthcheck.py agent/src/agent/__main__.py agent/tests/test_scheduler.py
git commit -m "feat(agent): scheduler, healthcheck, and container entrypoint"

---

Task 7: Verify CLI

Files:

• Create: agent/src/agent/verify.py
• Test: agent/tests/test_verify.py

Interfaces:

• Consumes: everything above.

• Produces: python -m agent.verify — post-deploy check printing: migrations applied + last sync runs (from agentdb), Firefly version, and the list of asset account names. Internal function verify(cfg: AgentConfig, client: FireflyClient | None = None) -> list[str] returns the report lines (printing wrapper around it); client param for tests.

• Step 1: Write the failing test — agent/tests/test_verify.py

from agent.db import ensure_database, run_migrations
from agent.jobs.heartbeat import run_heartbeat
from agent.verify import verify


class FakeFirefly:
    def about(self) -> dict:
        return {"version": "6.1.1"}

    def accounts(self, type: str = "asset") -> list[dict]:
        return [
            {"id": "1", "attributes": {"name": "NFCU Checking"}},
            {"id": "2", "attributes": {"name": "Venmo"}},
        ]


def test_verify_reports_db_firefly_and_accounts(cfg):
    ensure_database(cfg)
    run_migrations(cfg)
    run_heartbeat(cfg, client=FakeFirefly())

    report = "\n".join(verify(cfg, client=FakeFirefly()))

    assert "1 migration(s) applied" in report
    assert "heartbeat ok" in report
    assert "firefly: OK (version 6.1.1)" in report
    assert "2 asset account(s)" in report
    assert "NFCU Checking" in report

• Step 2: Run test to verify it fails

Run: cd agent && pytest tests/test_verify.py -v — Expected: FAIL with ModuleNotFoundError: No module named 'agent.verify'

• Step 3: Implement — agent/src/agent/verify.py

"""Post-deploy verification: `python -m agent.verify`."""

from __future__ import annotations

from agent.config import AgentConfig, load_config
from agent.db import connect
from agent.firefly import FireflyClient


def verify(cfg: AgentConfig, client: FireflyClient | None = None) -> list[str]:
    lines: list[str] = []

    with connect(cfg) as conn:
        migrations = conn.execute("SELECT count(*) FROM schema_migrations").fetchone()[0]
        runs = conn.execute(
            "SELECT connector, status, finished_at"
            " FROM sync_runs ORDER BY id DESC LIMIT 3"
        ).fetchall()
    lines.append(f"agentdb: OK ({migrations} migration(s) applied)")
    for connector, status, finished_at in runs:
        lines.append(f"  last run: {connector} {status} at {finished_at}")

    if client is None:
        client = FireflyClient(cfg.firefly_url, cfg.firefly_token)
    info = client.about()
    lines.append(f"firefly: OK (version {info.get('version', 'unknown')})")

    accounts = client.accounts()
    lines.append(f"firefly: {len(accounts)} asset account(s):")
    for account in accounts:
        lines.append(f"  - {account['attributes']['name']}")
    return lines


def main() -> int:
    for line in verify(load_config()):
        print(line)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

• Step 4: Run test to verify it passes

Run: cd agent && pytest tests/test_verify.py -v — Expected: PASS

• Step 5: Commit

git add agent/src/agent/verify.py agent/tests/test_verify.py
git commit -m "feat(agent): post-deploy verify command"

---

Task 8: Dockerfile, compose service, env docs, runbook

Files:

• Create: agent/Dockerfile, agent/.dockerignore, agent/README.md
• Modify: docker-compose.yml (add agent service after importer), .env.example (new section), README.md (repo root, one line)

Interfaces:

• Consumes: python -m agent and python -m agent.healthcheck from Task 6.

• Produces: docker compose build agent succeeds; docker compose config -q validates; .env.example documents the four new variables.

• Step 1: Create agent/.dockerignore

tests/
__pycache__/
*.egg-info/
.pytest_cache/

• Step 2: Create agent/Dockerfile

FROM python:3.12-slim

WORKDIR /app

COPY pyproject.toml ./
COPY src ./src

RUN pip install --no-cache-dir .

HEALTHCHECK --interval=60s --timeout=10s --start-period=30s \
    CMD ["python", "-m", "agent.healthcheck"]

CMD ["python", "-m", "agent"]

• Step 3: Build the image to verify

Run: docker build -t firefly-agent ./agent — Expected: builds successfully.

• Step 4: Add the agent service to docker-compose.yml

Insert after the importer service, matching the file's existing style:

  agent:
    build: ./agent
    restart: unless-stopped
    env_file: .env
    networks:
      - firefly
    depends_on:
      db:
        condition: service_healthy
      app:
        condition: service_healthy

Note: no Traefik labels and no traefik_proxy network — the agent is internal-only (Global Constraints).

• Step 5: Add the agent section to .env.example

Append:

# --- Accounting Agent ---
# Firefly API base URL as seen from inside the compose network
FIREFLY_API_URL=http://app:8080
# Personal Access Token: Firefly UI -> Options -> Profile -> OAuth -> Personal Access Tokens
AGENT_FIREFLY_TOKEN=REPLACE_WITH_PERSONAL_ACCESS_TOKEN
# Name of the agent's own database inside the existing Postgres container
AGENT_DB_NAME=agentdb
HEARTBEAT_INTERVAL_MINUTES=60

• Step 6: Validate compose config

Run: docker compose config -q — Expected: exit 0, no output. (Requires the local .env to exist; placeholder values are fine.)

• Step 7: Write agent/README.md

# firefly-agent

Worker container for the personal accounting agent. See the architecture spec:
`docs/superpowers/specs/2026-08-23-accounting-agent-architecture-design.md`.

## What it does (Phase 1)

On startup: creates `agentdb` in the stack's Postgres if missing, applies SQL
migrations from `src/agent/migrations/`, runs one heartbeat, then schedules the
heartbeat every `HEARTBEAT_INTERVAL_MINUTES`. The heartbeat calls Firefly's
`/api/v1/about` and records the outcome in `sync_runs`.

## Development

    python3 -m venv .venv && source .venv/bin/activate
    pip install -e './agent[dev]'
    cd agent && pytest          # needs Docker running (testcontainers)

## Configuration

All via environment variables (see `.env.example`, "Accounting Agent" section):
`FIREFLY_API_URL`, `AGENT_FIREFLY_TOKEN` (required), `AGENT_DB_NAME`,
`HEARTBEAT_INTERVAL_MINUTES`, plus the existing `DB_HOST`/`DB_PORT`/
`POSTGRES_USER`/`POSTGRES_PASSWORD`.

## Deploy & verify

    # on the host, in the repo directory
    git pull
    docker compose build agent
    docker compose up -d agent
    docker compose exec agent python -m agent.verify

Expected verify output: agentdb OK with migrations applied, recent heartbeat
runs, Firefly version, and your asset accounts listed.

Inspect heartbeats directly:

    docker compose exec db psql -U firefly -d agentdb \
      -c "SELECT connector, status, finished_at FROM sync_runs ORDER BY id DESC LIMIT 5;"

• Step 8: Add one line to the repo-root README.md

# Firefly III - jbnel.dev Finance Stack

`agent/` — personal accounting agent worker; see `agent/README.md` and `docs/superpowers/specs/`.

• Step 9: Run the full test suite one more time

Run: cd agent && pytest -v — Expected: all PASS

• Step 10: Commit

git add agent/Dockerfile agent/.dockerignore agent/README.md docker-compose.yml .env.example README.md
git commit -m "feat(agent): containerize agent and wire into compose stack"

---

Task 9: Production setup and deploy (human-in-the-loop with Jake)

No new code. This task configures Firefly and deploys the agent. Steps marked [Jake] need his hands (credentials, UI clicks); the rest can be driven together over SSH once the workflow is confirmed.

• Step 1: [Jake] Confirm the deploy workflow

The plan assumes: SSH to the host running the stack, git pull in the repo directory, docker compose commands there. Confirm or correct; record the actual workflow in agent/README.md if it differs.

• Step 2: [Jake] Create the Personal Access Token

Firefly UI (https://firefly.jbnel.dev) → Options → Profile → OAuth → Personal Access Tokens → Create new token, name it agent. Copy the token (shown once).

• Step 3: [Jake] Create asset accounts in Firefly

Firefly UI → Accounts → Asset accounts → Create. One per real account; suggested starting set (adjust to reality): NFCU Checking, NFCU Savings, TFCU Checking, Cash App, Venmo, PayPal. Opening balances optional — Phase 2 ingestion will populate transactions.

• Step 4: [Jake] Add agent variables to the host .env

Append the "Accounting Agent" section from .env.example to the host's .env, with the real token in AGENT_FIREFLY_TOKEN.

• Step 5: Deploy

On the host, in the repo directory:

git pull
docker compose build agent
docker compose up -d agent
docker compose ps agent   # wait for status: healthy

• Step 6: Verify

docker compose exec agent python -m agent.verify

Expected: agentdb: OK (1 migration(s) applied), a heartbeat ok run, firefly: OK (version 6.x), and the asset accounts from Step 3 listed. If firefly: shows 401, the token is wrong; if connection refused, check FIREFLY_API_URL=http://app:8080.

• Step 7: Confirm the schedule is live

After one heartbeat interval (default 60 min — or set HEARTBEAT_INTERVAL_MINUTES=2 temporarily and docker compose up -d agent to recreate):

docker compose exec db psql -U firefly -d agentdb \
  -c "SELECT connector, status, finished_at FROM sync_runs ORDER BY id DESC LIMIT 5;"

Expected: multiple heartbeat | ok rows with advancing timestamps. Restore the interval if it was lowered.

• Step 8: Close out Phase 1

Mark this plan's checkboxes done and commit any runbook corrections discovered during deploy:

git add -A && git commit -m "docs(agent): record actual deploy workflow from first production deploy"

Phase 1's definition of done: the agent container runs healthy on the host, heartbeats accumulate in sync_runs, and python -m agent.verify passes against production Firefly with real asset accounts listed.