Overview

adr/0001-import-baseline-and-goals.md

# ADR-0001: Import the Pass-1 baseline under a green suite; re-engineering goals
 
**Status**: Accepted
**Date**: 2026-07-19
 
## Context
 
Pass 1's `capstone-first-working-software` shipped a working, tested, hardened Habit Tracker
API. It is functionally complete but was built to prove Pass 1's promise (a small working
application), not to demonstrate Pass 2's promise (professional-grade internal structure,
performance discipline, and delivery workflow). Pass 2 taught SOLID + patterns (topic 21),
deliberate paradigm choice + functional core (22/23), safe concurrency (24), algorithmic
complexity (25), `EXPLAIN`-driven SQL tuning (26), and engineering workflow discipline (30),
framed by product/delivery judgment (32/33). This capstone re-engineers the Pass-1 app to
apply all of it, without changing its externally observable behavior.
 
## Decision
 
Import the Pass-1 app's baseline (`domain.py`, `models.py`, `auth.py`, `middleware.py`,
`schema_v1.sql`, `migration_v2.sql`) largely unchanged, confirm its test suite is green against
the imported baseline, then evolve it in three further ordered steps (SOLID/functional-core
refactor; concurrency + SQL tuning; workflow wrapping), each verified independently before the
next begins.
 
## Consequences
 
- **Positive**: every step has an independently green checkpoint (TDD-style: refactor, verify,
  proceed), so a regression is caught at the step that introduced it, not at the end.
- **Positive**: `auth.py` and `middleware.py` ship byte-identical to Pass 1 -- their correctness
  was already proven there; re-deriving them here would add risk for no benefit.
- **Trade-off**: this ADR set documents DECISIONS, not a full design document -- it is scoped to
  what changed and why, matching topic 30's "ADRs record a decision and its context, not an
  exhaustive spec."
 
## Verification
 
The Pass-1 app's OWN, unmodified test suite runs green against the imported baseline before any
Step-2 refactor begins: 34 tests passed (9 unit, 2 property, 23 integration -- the same count
Pass 1's own page documents), `ruff check`/`ruff format --check` clean, `pyright --strict` 0
errors, `pip-audit -l` clean (see this page's Step 1 transcript). The suite grows to 63 tests
only later, as Step 2 adds `test_services.py` and Step 3 adds concurrency/recent-activity
coverage -- this ADR's own scope is the baseline import alone, not the finished suite.

$ cd capstone-first-working-software/code && export CAPSTONE1_DB_PATH=/tmp/baseline.db \
  CAPSTONE1_AUTH_SECRET=verify-not-a-real-secret && python3 -m pytest -q
..................................                                       [100%]
34 passed in 2.46s

app/ports.py

"""capstone-solid-core: the repository PORT (topic 21 SOLID -- Dependency Inversion Principle).
 
`HabitService` (services.py) depends ONLY on this `Protocol` (structural typing, PEP 544 --
stable in the Python standard library `typing` module since Python 3.8, no third-party
dependency) -- never on `sqlite3`, never on a concrete database driver. A new storage backend
is added later by writing a new class shaped like this Protocol, with ZERO edits to
`HabitService` or to this file (Open/Closed Principle -- "open for extension, closed for
modification"). `test_services.py`'s `InMemoryHabitRepository` is exactly that new variation,
and it proves the point: it satisfies this same Protocol, `HabitService` never imports it, and
every existing test keeps passing unmodified.
"""
 
from __future__ import annotations
 
from typing import Protocol
 
from .domain import Habit
from .models import HabitCreate
 
 
class HabitRepository(Protocol):
    """The shape any storage backend must satisfy. Pure interface: no implementation, no
    state -- `Protocol` classes are never instantiated directly."""
 
    def create_habit(self, user_id: int, data: HabitCreate) -> Habit: ...
 
    def get_habit(self, habit_id: int, user_id: int) -> Habit | None: ...
 
    def list_habits(
        self, user_id: int, include_archived: bool = False
    ) -> list[Habit]: ...
 
    def search_habits(self, user_id: int, q: str) -> list[Habit]: ...
 
    def archive_habit(self, habit_id: int, user_id: int) -> Habit | None: ...
 
    def delete_habit(self, habit_id: int, user_id: int) -> bool: ...
 
    def record_checkin(
        self, habit_id: int, user_id: int, checkin_date_iso: str
    ) -> None: ...
 
    def recent_activity(self, user_id: int, limit: int) -> list[tuple[int, str]]: ...

app/services.py

"""capstone-solid-core: the application/use-case layer (topic 21 SOLID -- Single
Responsibility Principle). `HabitService` knows the HABIT-TRACKING RULES (e.g. "a check-in
needs an existing, owned habit") and orchestrates the repository port + the functional core --
it never touches `sqlite3`, never touches FastAPI's `Request`/`Response`. `main.py`'s route
handlers (the imperative shell's HTTP edge) are thin: parse the request, call one
`HabitService` method, shape the response -- no business rule lives in a route handler.
 
This is exactly the layer `test_services.py` exercises directly, with a fake in-memory
repository and zero real database -- proof that the business rules are independently testable
from the storage mechanism (Dependency Inversion, topic 21).
"""
 
from __future__ import annotations
 
from datetime import date
 
from .domain import Habit
from .models import HabitCreate
from .ports import HabitRepository
 
 
class HabitService:
    def __init__(self, repo: HabitRepository) -> None:
        self._repo = (
            repo  # => depends on the ABSTRACTION (DIP), never a concrete DB class
        )
 
    def create_habit(self, user_id: int, data: HabitCreate) -> Habit:
        return self._repo.create_habit(user_id, data)
 
    def get_habit(self, habit_id: int, user_id: int) -> Habit | None:
        return self._repo.get_habit(habit_id, user_id)
 
    def list_habits(
        self, user_id: int, include_archived: bool = False, q: str | None = None
    ) -> list[Habit]:
        if q:  # => the ONE business rule this method owns: q present means SEARCH, not LIST
            return self._repo.search_habits(user_id, q)
        return self._repo.list_habits(user_id, include_archived)
 
    def archive_habit(self, habit_id: int, user_id: int) -> Habit | None:
        return self._repo.archive_habit(habit_id, user_id)
 
    def delete_habit(self, habit_id: int, user_id: int) -> bool:
        return self._repo.delete_habit(habit_id, user_id)
 
    def record_checkin(
        self, habit_id: int, user_id: int, checkin_day: date
    ) -> Habit | None:
        """The business rule this method owns: a check-in is only valid against a habit this
        user actually owns -- the ownership check happens HERE, once, rather than being
        re-implemented by every caller."""
        habit = self._repo.get_habit(habit_id, user_id)
        if habit is None:
            return None
        self._repo.record_checkin(habit_id, user_id, checkin_day.isoformat())
        habit.record_checkin(
            checkin_day
        )  # => mirror the write into the in-memory object too --
        return (
            habit  # => the caller's response reflects it without a second DB round trip
        )
 
    def recent_activity(self, user_id: int, limit: int) -> list[tuple[int, str]]:
        return self._repo.recent_activity(user_id, limit)

app/repository_sqlite.py

"""capstone-solid-core: the ONE SQLite adapter (topic 10 SQL Essentials + topic 21 SOLID
Dependency Inversion) -- the imperative shell that actually talks to the database. Every
statement binds its inputs as `?` placeholders (co-20 parameterized-queries, carried forward
unchanged from Pass 1) -- never an f-string -- so user-controlled data is always treated as
DATA, never as SQL text. `init_db` applies the base schema, then two additive migrations
(schema-migration, topic 10 co-22), tracked by `PRAGMA user_version`, so calling it twice
against an already-migrated file is a safe no-op.
 
`SqliteHabitRepository` is the ONLY class in this codebase that satisfies `ports.HabitRepository`
that ships in production -- `HabitService` (services.py) never imports this module by name; it
only imports the `HabitRepository` Protocol from ports.py (Dependency Inversion Principle, topic
21). `test_services.py`'s `InMemoryHabitRepository` is a SECOND class satisfying the same
Protocol, added with zero edits here (Open/Closed Principle).
"""
 
from __future__ import annotations
 
import sqlite3
from datetime import date
from pathlib import Path
 
from .domain import Habit
from .models import HabitCreate, UserPublic
 
SCHEMA_V1_PATH = Path(__file__).parent / "schema_v1.sql"
MIGRATION_V2_PATH = Path(__file__).parent / "migration_v2.sql"
MIGRATION_V3_PATH = Path(__file__).parent / "migration_v3.sql"
 
 
def get_connection(db_path: str) -> sqlite3.Connection:
    conn = sqlite3.connect(db_path)
    conn.row_factory = (
        sqlite3.Row
    )  # => rows addressable by column name, not just position
    conn.execute(
        "PRAGMA foreign_keys = ON"
    )  # => enforce FK constraints on this connection
    conn.execute(
        "PRAGMA journal_mode = WAL"
    )  # => co-XX (topic 24): Write-Ahead Logging lets one writer and MANY concurrent
    # => readers proceed without blocking each other -- Step 3's concurrent digest reads
    # => from several separate connections/processes WHILE the app keeps serving writes.
    return conn
 
 
def init_db(
    db_path: str,
) -> None:  # => topic 10 co-22 schema-migration: safe to call repeatedly
    conn = get_connection(db_path)
    current_version = int(conn.execute("PRAGMA user_version").fetchone()[0])
    if current_version < 1:  # => base schema not yet applied
        conn.executescript(SCHEMA_V1_PATH.read_text(encoding="utf-8"))
        conn.execute("PRAGMA user_version = 1")
        current_version = 1
    if current_version < 2:  # => additive migration (habits.archived) not yet applied
        conn.executescript(MIGRATION_V2_PATH.read_text(encoding="utf-8"))
        conn.execute("PRAGMA user_version = 2")
        current_version = 2
    if (
        current_version < 3
    ):  # => Step 3's composite index (co-XX, topic 26 EXPLAIN-guided)
        conn.executescript(MIGRATION_V3_PATH.read_text(encoding="utf-8"))
        conn.execute("PRAGMA user_version = 3")
    conn.commit()
    conn.close()
 
 
def ping(
    conn: sqlite3.Connection,
) -> bool:  # => the cheapest possible real query -- /ready
    conn.execute("SELECT 1")
    return True
 
 
# --- users -----------------------------------------------------------------------------------
 
 
def create_user(
    conn: sqlite3.Connection, username: str, password_hash: str
) -> UserPublic:
    cursor = conn.execute(
        "INSERT INTO users (username, password_hash) VALUES (?, ?)",  # => parameterized
        (username, password_hash),
    )
    conn.commit()
    row = conn.execute(
        "SELECT id, username, created_at FROM users WHERE id = ?", (cursor.lastrowid,)
    ).fetchone()
    assert row is not None  # => guaranteed by the INSERT above
    return UserPublic(
        id=int(row["id"]),
        username=str(row["username"]),
        created_at=str(row["created_at"]),
    )
 
 
def get_user_by_username(conn: sqlite3.Connection, username: str) -> sqlite3.Row | None:
    return conn.execute(
        "SELECT * FROM users WHERE username = ?", (username,)
    ).fetchone()
 
 
# --- habits + check-ins: the class satisfying ports.HabitRepository ---------------------------
 
 
def _load_habit(conn: sqlite3.Connection, habit_row: sqlite3.Row) -> Habit:
    """Build a `Habit` domain object from a `habits` row plus its `checkins` rows, replaying
    every stored check-in through `record_checkin` -- the DB is the source of truth; the domain
    object is a transient in-memory VIEW of it, rebuilt on every load (DD-33 taming-state)."""
    habit = Habit(
        id=int(habit_row["id"]),
        name=str(habit_row["name"]),
        archived=bool(habit_row["archived"]),
    )
    for checkin_row in conn.execute(
        "SELECT checkin_date FROM checkins WHERE habit_id = ?", (habit_row["id"],)
    ):
        habit.record_checkin(date.fromisoformat(str(checkin_row["checkin_date"])))
    return habit
 
 
class SqliteHabitRepository:
    """Implements `ports.HabitRepository` against a `sqlite3.Connection`. Structural typing
    (PEP 544) means this class need not, and does not, inherit from `HabitRepository` --
    Python's `Protocol` matches by SHAPE, not by declared ancestry."""
 
    def __init__(self, conn: sqlite3.Connection) -> None:
        self._conn = conn
 
    def create_habit(self, user_id: int, data: HabitCreate) -> Habit:
        cursor = self._conn.execute(
            "INSERT INTO habits (user_id, name) VALUES (?, ?)", (user_id, data.name)
        )
        self._conn.commit()
        row = self._conn.execute(
            "SELECT * FROM habits WHERE id = ?", (cursor.lastrowid,)
        ).fetchone()
        assert row is not None
        return _load_habit(self._conn, row)
 
    def get_habit(self, habit_id: int, user_id: int) -> Habit | None:
        row = self._conn.execute(
            "SELECT * FROM habits WHERE id = ? AND user_id = ?", (habit_id, user_id)
        ).fetchone()
        return _load_habit(self._conn, row) if row is not None else None
 
    def list_habits(self, user_id: int, include_archived: bool = False) -> list[Habit]:
        if include_archived:
            rows = self._conn.execute(
                "SELECT * FROM habits WHERE user_id = ? ORDER BY id", (user_id,)
            ).fetchall()
        else:
            rows = self._conn.execute(
                "SELECT * FROM habits WHERE user_id = ? AND archived = 0 ORDER BY id",
                (user_id,),
            ).fetchall()
        return [_load_habit(self._conn, row) for row in rows]
 
    def search_habits(self, user_id: int, q: str) -> list[Habit]:
        rows = self._conn.execute(
            "SELECT * FROM habits WHERE user_id = ? AND name LIKE '%' || ? || '%' ORDER BY id",
            (user_id, q),  # => `?`: q is DATA, never spliced into the SQL text
        ).fetchall()
        return [_load_habit(self._conn, row) for row in rows]
 
    def archive_habit(self, habit_id: int, user_id: int) -> Habit | None:
        row = self._conn.execute(
            "SELECT * FROM habits WHERE id = ? AND user_id = ?", (habit_id, user_id)
        ).fetchone()
        if row is None:
            return None
        habit = _load_habit(self._conn, row)
        habit.archive()  # => the ONE guarded mutator (topic 08), not a hand-written flag flip
        self._conn.execute(
            "UPDATE habits SET archived = ? WHERE id = ? AND user_id = ?",
            (int(habit.archived), habit_id, user_id),
        )
        self._conn.commit()
        return habit
 
    def delete_habit(self, habit_id: int, user_id: int) -> bool:
        cursor = self._conn.execute(
            "DELETE FROM habits WHERE id = ? AND user_id = ?", (habit_id, user_id)
        )
        self._conn.commit()
        return cursor.rowcount > 0
 
    def record_checkin(
        self, habit_id: int, user_id: int, checkin_date_iso: str
    ) -> None:
        self._conn.execute(
            "INSERT INTO checkins (habit_id, user_id, checkin_date) VALUES (?, ?, ?) "
            "ON CONFLICT (habit_id, checkin_date) DO NOTHING",  # => a repeat check-in is a no-op
            (habit_id, user_id, checkin_date_iso),
            # => `user_id` is written HERE too (Step 3's denormalization, migration_v3.sql) --
            # => the ONE place this app ever creates a checkins row, so the copy can never drift
        )
        self._conn.commit()
 
    def recent_activity(self, user_id: int, limit: int) -> list[tuple[int, str]]:
        """Step 3's EXPLAIN-guided-index query (topic 26 co-XX denormalization-tradeoffs): this
        user's most recent check-ins across EVERY habit, newest first -- no join, thanks to
        migration_v3.sql's denormalized `checkins.user_id` + composite index."""
        rows = self._conn.execute(
            "SELECT habit_id, checkin_date FROM checkins "
            "WHERE user_id = ? ORDER BY checkin_date DESC LIMIT ?",
            (user_id, limit),
        ).fetchall()
        return [(int(row["habit_id"]), str(row["checkin_date"])) for row in rows]

app/domain.py

"""capstone-solid-core: the functional core (topic 22 Programming Paradigms, topic 23
Functional Programming) plus the thin imperative-shell object it backs (topic 08 OOP, topic 21
SOLID). Every computation below is a PURE function -- no I/O, no mutation of anything outside
its own arguments, same output for the same input every time -- so each one is independently
unit-testable and benchmarkable with zero database and zero HTTP server (functional core /
imperative shell, DD-33 taming-state). `Habit` is the imperative shell: it owns identity and
mutable state (its own check-in set), and delegates every actual computation to the pure
functions below instead of inlining the logic as methods, the same "bundle state with the
operations that guard it" idea Pass 1 topic 08 taught, now split so the REASONING part carries
no state at all.
"""
 
from __future__ import annotations
 
from dataclasses import dataclass, field
from datetime import date, timedelta
 
 
def current_streak(checkin_dates: set[date], today: date) -> int:
    """PURE (topic 23): the count of CONSECUTIVE days, walking backward from `today`, present
    in `checkin_dates`. Unchanged from Pass 1's algorithm (topic 07 co-09 hash-set: O(1)
    average membership per day) -- now extracted as a free function so it can be called,
    tested, and benchmarked without ever constructing a `Habit`."""
    streak = 0
    day = today
    while day in checkin_dates:  # => O(1) average hash-set membership test (topic 07)
        streak += 1
        day -= timedelta(days=1)
    return streak  # => O(streak) total, same bound Pass 1 established
 
 
def longest_streak_ever_naive(checkin_dates: set[date]) -> int:
    """BEFORE (topic 25 Advanced Algorithms baseline, kept here for the benchmark's own
    comparison -- never called by the shipped app): sort every date, then scan once for the
    longest run of consecutive days. `sorted()` costs O(n log n) (Timsort) and DOMINATES the
    total cost for a large check-in history -- the O(n) scan that follows it is comparatively
    free."""
    if not checkin_dates:
        return 0
    ordered = sorted(
        checkin_dates
    )  # => O(n log n): the cost this function exists to remove
    longest = 1
    current = 1
    for previous_day, this_day in zip(ordered, ordered[1:]):
        if this_day - previous_day == timedelta(days=1):  # => consecutive calendar days
            current += 1
            longest = max(longest, current)
        else:
            current = 1  # => the run broke; restart counting from this_day
    return longest
 
 
def longest_streak_ever(checkin_dates: set[date]) -> int:
    """AFTER (topic 25 Advanced Algorithms): O(n) total, no sort. The classic "longest
    consecutive sequence over a hash-set" technique applied to calendar dates instead of
    integers: a date only STARTS counting a run if the day before it is absent from the set --
    every other date is skipped in O(1) by that one membership test. Because only run-STARTS
    trigger the inner walk, and every date is visited by at most one run's walk across the
    WHOLE function, the total work across all iterations is O(n), not O(n^2) and not
    O(n log n) -- this is an amortized argument (topic 19 CS Foundations complexity
    reasoning), not a per-call bound on the inner while loop alone.
 
    IMPLEMENTATION NOTE (measured, not assumed -- see bench/benchmark_algorithm.py): a FIRST
    version of this function worked directly with `date`/`timedelta` objects (`day -
    timedelta(days=1) in checkin_dates`, `probe += timedelta(days=1)`) and was actually SLOWER
    in wall-clock terms than the O(n log n) `_naive` baseline above at every size tested, up to
    n=500,000 -- Python's built-in `sorted()` is implemented in C with very low per-comparison
    overhead, while constructing a new `timedelta` object on every iteration of a pure-Python
    loop is comparatively expensive. Converting each `date` to its integer ordinal
    (`date.toordinal()`, Python stdlib -- docs.python.org/3/library/datetime.html:
    "Return the proleptic Gregorian ordinal of the date") once up front, then doing the SAME
    algorithm over a `set[int]` with plain integer +/-1 instead of `timedelta` construction,
    measured 2.59x-2.86x FASTER than the naive baseline across the same size range -- the
    asymptotic O(n) advantage only became a real wall-clock advantage once the Python-level
    constant-factor overhead (repeated object construction) was removed. Big-O describes what
    happens as n grows; it does not by itself guarantee a win at any one concrete n in a
    language with per-object overhead -- that has to be measured."""
    if not checkin_dates:
        return 0
    ordinals = {
        day.toordinal() for day in checkin_dates
    }  # => O(n): once, not per-comparison
    longest = 0
    for ordinal in ordinals:  # => O(n) over the set, one membership test per element
        if (ordinal - 1) in ordinals:
            continue  # => NOT a run-start -- some other day's walk already counts this one
        run_length = 1
        probe = ordinal + 1
        while probe in ordinals:  # => walks forward ONLY from a genuine run-start
            run_length += 1
            probe += 1
        longest = max(longest, run_length)
    return longest
 
 
@dataclass(slots=True)  # => slots=True (topic 08 co-06): no per-instance __dict__
class Habit:
    """The imperative shell (DD-33 taming-state): owns identity (`id`, `name`, `archived`) and
    the ONE piece of mutable state (`_checkin_dates`), and is the ONLY thing anything outside
    this module ever touches directly. Every computation is DELEGATED to the pure functions
    above, not inlined -- so `current_streak`/`longest_streak_ever` never need a `Habit`
    instance to be tested or benchmarked, and this class never needs to know HOW they work."""
 
    id: int  # => the DB row id; 0 for a not-yet-persisted (in-memory only) instance
    name: str  # => validated non-blank in __post_init__ below
    archived: bool = (
        False  # => guarded ONLY by archive() below -- never flipped from outside
    )
    _checkin_dates: set[date] = field(
        default_factory=set[date]
    )  # => co-09: hash-set, O(1) average membership
 
    def __post_init__(self) -> None:
        if not self.name.strip():
            raise ValueError("habit name must not be blank")
 
    def archive(self) -> None:
        """The ONE way to archive a habit -- archiving twice is a safe no-op, never an error."""
        self.archived = True
 
    def record_checkin(self, day: date) -> None:
        """Record a check-in for `day`. Idempotent: `set.add` on an already-present element
        changes nothing."""
        self._checkin_dates.add(day)  # => O(1) average insert into the hash-set
 
    def has_checkin_on(self, day: date) -> bool:
        return day in self._checkin_dates  # => O(1) average
 
    def checkin_count(self) -> int:
        return len(self._checkin_dates)
 
    def current_streak(self, today: date) -> int:
        """Delegates to the pure `current_streak` function above -- this method adds no logic
        of its own, only the state to compute over."""
        return current_streak(self._checkin_dates, today)
 
    def longest_streak_ever(self) -> int:
        """Delegates to the pure, O(n) `longest_streak_ever` function above."""
        return longest_streak_ever(self._checkin_dates)

app/main.py

"""capstone-solid-core: the re-engineered Habit Tracker API -- the imperative shell (topic
21 SOLID + topic 22/23 functional core / imperative shell split) that wires the pure functional
core (domain.py), the port + adapter (ports.py / repository_sqlite.py), and the application
layer (services.py) into a running FastAPI app. Route handlers below are deliberately thin:
parse the request, call ONE `HabitService` method, shape the response -- no business rule lives
here (Single Responsibility, topic 21). Auth (auth.py) and the two middlewares (middleware.py)
are reused BYTE-IDENTICAL from the Pass-1 capstone -- proven correct there, unchanged here.
 
Run with: uvicorn app.main:app --port 8101
"""
 
import os
import sqlite3
from collections.abc import Iterator
from datetime import date
 
from fastapi import Depends, FastAPI, HTTPException, Query, Request, Response
from fastapi.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
 
from . import auth
from . import repository_sqlite as repo
from .domain import Habit
from .middleware import make_token_check_middleware, security_headers_middleware
from .models import (
    CheckinCreate,
    CheckinPublic,
    HabitCreate,
    HabitPublic,
    RecentCheckinPublic,
    TokenResponse,
    UserLogin,
    UserPublic,
    UserRegister,
)
from .services import HabitService
 
# overridable so tests can point at a fresh, isolated file
DB_PATH = os.environ.get(
    "CAPSTONE_SOLID_CORE_DB_PATH", os.path.join(os.path.dirname(__file__), "habits.db")
)
# REQUIRED, no hardcoded fallback -- refuses to start if the secret is missing, by design
AUTH_SECRET = os.environ["CAPSTONE_SOLID_CORE_AUTH_SECRET"]
 
repo.init_db(
    DB_PATH
)  # => applies schema_v1 + migration_v2 + migration_v3 once, at startup
 
app = FastAPI(title="capstone-solid-core: Habit Tracker API (Pass-2 professional core)")
 
 
def get_db() -> Iterator[sqlite3.Connection]:  # => one connection per request
    conn = repo.get_connection(DB_PATH)
    try:
        yield conn
    finally:
        conn.close()
 
 
def get_habit_service(
    conn: sqlite3.Connection = Depends(get_db),
) -> HabitService:
    """Composition root (topic 21 DIP): the ONE place a concrete `SqliteHabitRepository` is
    constructed and handed to `HabitService` as its abstract `HabitRepository` port. Every
    route handler below depends on `HabitService`, never on `repo`/`sqlite3` directly."""
    return HabitService(repo.SqliteHabitRepository(conn))
 
 
def _resolve_token(token: str) -> int | None:  # => the ONE place AUTH_SECRET is used
    return auth.resolve_token(token, AUTH_SECRET)
 
 
def _current_user_id(request: Request) -> int:
    user_id = getattr(
        request.state, "user_id", None
    )  # => set by the token-check middleware
    assert user_id is not None  # => guaranteed once token_check_middleware has run
    return int(user_id)
 
 
app.middleware("http")(make_token_check_middleware(_resolve_token))  # guards /habits
app.middleware("http")(security_headers_middleware)  # stamps every response
 
 
@app.exception_handler(
    StarletteHTTPException
)  # => registered on Starlette's BASE class
async def handle_http_exception(
    request: Request, exc: StarletteHTTPException
) -> JSONResponse:
    # => fastapi.HTTPException is a SUBCLASS of Starlette's, so this one handler also catches
    # => exceptions Starlette raises internally (unmatched-route 404s, wrong-method 405s)
    body = (
        exc.detail
        if isinstance(exc.detail, dict)
        else {"error": {"code": "error", "message": str(exc.detail)}}
    )
    return JSONResponse(status_code=exc.status_code, content=body)
 
 
def _habit_to_public(conn: sqlite3.Connection, habit: Habit) -> HabitPublic:
    """One shared place that turns a domain `Habit` into the HTTP response shape."""
    row = conn.execute(
        "SELECT created_at FROM habits WHERE id = ?", (habit.id,)
    ).fetchone()
    assert row is not None  # => the caller only ever passes a habit it just loaded
    return HabitPublic(
        id=habit.id,
        name=habit.name,
        created_at=str(row["created_at"]),
        archived=habit.archived,
        checkin_count=habit.checkin_count(),
        current_streak=habit.current_streak(date.today()),
    )
 
 
@app.get("/health")  # => LIVENESS -- always 200, no DB dependency at all
def health() -> dict[str, str]:
    return {"status": "ok"}
 
 
@app.get("/ready")  # => READINESS -- genuinely pings the database
def ready(
    response: Response, conn: sqlite3.Connection = Depends(get_db)
) -> dict[str, str]:
    try:
        repo.ping(conn)
        return {"status": "ready"}
    except sqlite3.OperationalError as exc:
        response.status_code = 503
        return {"status": "not_ready", "reason": str(exc)}
 
 
@app.post("/auth/register", response_model=UserPublic, status_code=201)
def register_route(
    body: UserRegister, conn: sqlite3.Connection = Depends(get_db)
) -> UserPublic:
    existing = repo.get_user_by_username(conn, body.username)
    if existing is not None:
        raise HTTPException(
            status_code=409,
            detail={"error": {"code": "conflict", "message": "username already taken"}},
        )
    password_hash = auth.hash_password(
        body.password
    )  # => hash BEFORE it ever touches the DB
    return repo.create_user(conn, body.username, password_hash)
 
 
@app.post("/auth/login", response_model=TokenResponse)
def login_route(
    body: UserLogin, conn: sqlite3.Connection = Depends(get_db)
) -> TokenResponse:
    row = repo.get_user_by_username(conn, body.username)
    generic_error = (
        HTTPException(  # => SAME message for "no such user" and "wrong password"
            status_code=401,
            detail={
                "error": {
                    "code": "unauthorized",
                    "message": "invalid username or password",
                }
            },
        )
    )
    if row is None:
        raise generic_error
    if not auth.verify_password(str(row["password_hash"]), body.password):
        raise generic_error
    token = auth.issue_token(int(row["id"]), AUTH_SECRET)
    return TokenResponse(access_token=token)
 
 
# guarded: token required
@app.post("/habits", response_model=HabitPublic, status_code=201)
def create_habit_route(
    body: HabitCreate,
    request: Request,
    conn: sqlite3.Connection = Depends(get_db),
    service: HabitService = Depends(get_habit_service),
) -> HabitPublic:
    user_id = _current_user_id(request)
    habit = service.create_habit(user_id, body)
    return _habit_to_public(conn, habit)
 
 
# guarded -- token required (reads are user-scoped)
@app.get("/habits", response_model=list[HabitPublic])
def list_habits_route(
    request: Request,
    q: str | None = Query(default=None, max_length=200),
    include_archived: bool = Query(default=False),
    conn: sqlite3.Connection = Depends(get_db),
    service: HabitService = Depends(get_habit_service),
) -> list[HabitPublic]:
    user_id = _current_user_id(request)
    habits = service.list_habits(user_id, include_archived, q)
    return [_habit_to_public(conn, h) for h in habits]
 
 
# guarded -- token required, ownership-scoped
@app.get("/habits/{habit_id}", response_model=HabitPublic)
def get_habit_route(
    habit_id: int,
    request: Request,
    conn: sqlite3.Connection = Depends(get_db),
    service: HabitService = Depends(get_habit_service),
) -> HabitPublic:
    user_id = _current_user_id(request)
    habit = service.get_habit(habit_id, user_id)
    if habit is None:
        raise HTTPException(
            status_code=404,
            detail={"error": {"code": "not_found", "message": "no such habit"}},
        )
    return _habit_to_public(conn, habit)
 
 
@app.post("/habits/{habit_id}/checkins", response_model=CheckinPublic, status_code=201)
def record_checkin_route(
    habit_id: int,
    body: CheckinCreate,
    request: Request,
    service: HabitService = Depends(get_habit_service),
) -> CheckinPublic:
    user_id = _current_user_id(request)
    checkin_day = body.checkin_date if body.checkin_date is not None else date.today()
    habit = service.record_checkin(habit_id, user_id, checkin_day)
    if habit is None:
        raise HTTPException(
            status_code=404,
            detail={"error": {"code": "not_found", "message": "no such habit"}},
        )
    return CheckinPublic(
        habit_id=habit_id,
        checkin_date=checkin_day,
        checkin_count=habit.checkin_count(),
        current_streak=habit.current_streak(date.today()),
    )
 
 
@app.post("/habits/{habit_id}/archive", response_model=HabitPublic)
def archive_habit_route(
    habit_id: int,
    request: Request,
    conn: sqlite3.Connection = Depends(get_db),
    service: HabitService = Depends(get_habit_service),
) -> HabitPublic:
    user_id = _current_user_id(request)
    habit = service.archive_habit(habit_id, user_id)
    if habit is None:
        raise HTTPException(
            status_code=404,
            detail={"error": {"code": "not_found", "message": "no such habit"}},
        )
    return _habit_to_public(conn, habit)
 
 
# guarded -- token required, ownership-scoped
@app.delete("/habits/{habit_id}", status_code=204)
def delete_habit_route(
    habit_id: int,
    request: Request,
    service: HabitService = Depends(get_habit_service),
) -> None:
    user_id = _current_user_id(request)
    if not service.delete_habit(habit_id, user_id):
        raise HTTPException(
            status_code=404,
            detail={"error": {"code": "not_found", "message": "no such habit"}},
        )
 
 
# Step 3's EXPLAIN-guided-index endpoint (topic 26) -- no join, no sort at the DB level; see
# migration_v3.sql + bench/explain_query_plan.sh for the real, captured before/after plan.
@app.get("/habits/activity/recent", response_model=list[RecentCheckinPublic])
def recent_activity_route(
    request: Request,
    limit: int = Query(default=20, ge=1, le=200),
    service: HabitService = Depends(get_habit_service),
) -> list[RecentCheckinPublic]:
    user_id = _current_user_id(request)
    rows = service.recent_activity(user_id, limit)
    return [RecentCheckinPublic(checkin_date=date.fromisoformat(d)) for _, d in rows]

app/models.py

"""capstone-solid-core: typed request/response models (topic 11 HTTP JSON API + topic 17 co-07
allow-list validation), reused unchanged from the Pass-1 capstone plus one addition for this
capstone's own new endpoint: `RecentCheckinPublic` (Step 3's EXPLAIN-guided-index endpoint).
`app/digest.py`'s own `HabitDigest` dataclass serves Step 3's concurrency story instead -- it is
never exposed over HTTP (see digest.py's own docstring for why), so it has no Pydantic response
model here. Every field FastAPI/Pydantic validates BEFORE any handler code runs -- a hostile or
malformed body never reaches the service/repository layers.
"""
 
from datetime import date
 
from pydantic import BaseModel, Field
 
# --- auth surface -- co-07 allow-list validation (topic 17) -------------------------------
 
 
class UserRegister(BaseModel):
    username: str = Field(min_length=3, max_length=32, pattern=r"^[a-zA-Z0-9_]+$")
    # => an allow-list: only letters/digits/underscore -- rejects `admin'--` outright,
    # => before any handler code runs (topic 17 co-07)
    password: str = Field(
        min_length=8, max_length=128
    )  # => bounds only -- the HASH is what protects it
 
 
class UserLogin(BaseModel):
    username: str = Field(min_length=3, max_length=32, pattern=r"^[a-zA-Z0-9_]+$")
    password: str = Field(min_length=8, max_length=128)
 
 
class UserPublic(BaseModel):
    id: int
    username: str
    created_at: str
 
 
class TokenResponse(BaseModel):
    access_token: str
    token_type: str = "bearer"
 
 
# --- habits + check-ins -- topic 11 CRUD + topic 10 normalized data -----------------------
 
 
class HabitCreate(BaseModel):
    name: str = Field(min_length=1, max_length=100)
 
 
class HabitPublic(BaseModel):
    id: int
    name: str
    created_at: str
    archived: bool
    checkin_count: int
    current_streak: (
        int  # => computed by app/domain.py's Habit.current_streak(), never stored
    )
 
 
class CheckinCreate(BaseModel):
    checkin_date: date | None = (
        None  # => omit to check in for TODAY; pass an explicit date to backfill
    )
 
 
class CheckinPublic(BaseModel):
    habit_id: int
    checkin_date: date
    checkin_count: int
    current_streak: int
 
 
# --- capstone-solid-core addition: recent check-ins (Step 3's SQL tuning) -----------------
 
 
class RecentCheckinPublic(BaseModel):
    checkin_date: date

tests/test_services.py

"""capstone-solid-core: tests for the application layer (topic 21 SOLID) -- AND the Open/Closed
Principle proof the syllabus spec requires: `InMemoryHabitRepository` below is a SECOND class
satisfying `ports.HabitRepository`, written ENTIRELY IN THIS TEST FILE. It required editing
NEITHER `app/services.py` NOR `app/ports.py` NOR `app/repository_sqlite.py` -- "closed for
modification" -- while `HabitService` accepts it exactly as it accepts the real SQLite adapter
-- "open for extension" (Python's `typing.Protocol`, PEP 544, matches by structural shape, not
by declared inheritance -- `InMemoryHabitRepository` never imports or subclasses
`HabitRepository`).
 
Every test below exercises `HabitService` with ZERO real database and ZERO HTTP server --
proof that the business rules in services.py are independently testable from the storage
mechanism (Dependency Inversion Principle).
"""
 
from __future__ import annotations
 
from datetime import date
 
import pytest
 
from app.domain import Habit
from app.models import HabitCreate
from app.services import HabitService
 
 
class InMemoryHabitRepository:
    """The NEW variation (topic 21 OCP): a dict-backed repository satisfying
    `ports.HabitRepository` by SHAPE alone. Never touches sqlite3. Never edits any shipped
    file. Exists ONLY in this test module."""
 
    def __init__(self) -> None:
        self._habits: dict[int, Habit] = {}
        self._owners: dict[int, int] = {}  # => habit_id -> user_id
        self._next_id = 1
 
    def create_habit(self, user_id: int, data: HabitCreate) -> Habit:
        habit = Habit(id=self._next_id, name=data.name)
        self._habits[habit.id] = habit
        self._owners[habit.id] = user_id
        self._next_id += 1
        return habit
 
    def get_habit(self, habit_id: int, user_id: int) -> Habit | None:
        if self._owners.get(habit_id) != user_id:
            return None
        return self._habits.get(habit_id)
 
    def list_habits(self, user_id: int, include_archived: bool = False) -> list[Habit]:
        return [
            h
            for hid, h in self._habits.items()
            if self._owners.get(hid) == user_id and (include_archived or not h.archived)
        ]
 
    def search_habits(self, user_id: int, q: str) -> list[Habit]:
        return [
            h
            for h in self.list_habits(user_id, include_archived=True)
            if q.lower() in h.name.lower()
        ]
 
    def archive_habit(self, habit_id: int, user_id: int) -> Habit | None:
        habit = self.get_habit(habit_id, user_id)
        if habit is None:
            return None
        habit.archive()
        return habit
 
    def delete_habit(self, habit_id: int, user_id: int) -> bool:
        if self._owners.get(habit_id) != user_id:
            return False
        del self._habits[habit_id]
        del self._owners[habit_id]
        return True
 
    def record_checkin(
        self, habit_id: int, user_id: int, checkin_date_iso: str
    ) -> None:
        habit = self._habits[habit_id]
        habit.record_checkin(date.fromisoformat(checkin_date_iso))
 
    def recent_activity(self, user_id: int, limit: int) -> list[tuple[int, str]]:
        entries: list[tuple[int, str]] = []
        for hid, habit in self._habits.items():
            if self._owners.get(hid) != user_id:
                continue
            for checkin_day in habit._checkin_dates:  # => test-only introspection
                entries.append((hid, checkin_day.isoformat()))
        entries.sort(key=lambda pair: pair[1], reverse=True)
        return entries[:limit]
 
 
@pytest.fixture()
def service() -> HabitService:
    return HabitService(
        InMemoryHabitRepository()
    )  # => the NEW variation, wired in directly
 
 
class TestHabitServiceWithInMemoryRepository:
    """Every assertion below is IDENTICAL in spirit to test_app.py's integration tests against
    the real SQLite-backed API -- proof the same business rules hold regardless of which
    HabitRepository implementation HabitService is handed (OCP + DIP working together)."""
 
    def test_create_then_get_round_trips(self, service: HabitService) -> None:
        created = service.create_habit(
            user_id=1, data=HabitCreate(name="Read 20 minutes")
        )
        fetched = service.get_habit(created.id, user_id=1)
        assert fetched is not None
        assert fetched.name == "Read 20 minutes"
 
    def test_checkin_updates_streak(self, service: HabitService) -> None:
        habit = service.create_habit(user_id=1, data=HabitCreate(name="Floss"))
        today = date(2026, 7, 16)
        updated = service.record_checkin(habit.id, user_id=1, checkin_day=today)
        assert updated is not None
        assert updated.current_streak(today) == 1
 
    def test_checkin_against_a_habit_the_caller_does_not_own_is_rejected(
        self, service: HabitService
    ) -> None:
        habit = service.create_habit(user_id=1, data=HabitCreate(name="Dave's habit"))
        result = service.record_checkin(
            habit.id, user_id=2, checkin_day=date(2026, 7, 16)
        )
        assert (
            result is None
        )  # => the SAME ownership rule test_app.py verifies over HTTP
 
    def test_search_filters_by_substring(self, service: HabitService) -> None:
        service.create_habit(user_id=1, data=HabitCreate(name="Read 20 minutes"))
        service.create_habit(user_id=1, data=HabitCreate(name="Drink water"))
        results = service.list_habits(user_id=1, q="Read")
        assert len(results) == 1
        assert results[0].name == "Read 20 minutes"
 
    def test_archiving_hides_from_the_default_list(self, service: HabitService) -> None:
        habit = service.create_habit(user_id=1, data=HabitCreate(name="Old habit"))
        service.archive_habit(habit.id, user_id=1)
        assert service.list_habits(user_id=1) == []
        assert len(service.list_habits(user_id=1, include_archived=True)) == 1
 
    def test_recent_activity_orders_newest_first(self, service: HabitService) -> None:
        habit = service.create_habit(
            user_id=1, data=HabitCreate(name="Read 20 minutes")
        )
        service.record_checkin(habit.id, user_id=1, checkin_day=date(2026, 7, 1))
        service.record_checkin(habit.id, user_id=1, checkin_day=date(2026, 7, 16))
        activity = service.recent_activity(user_id=1, limit=10)
        assert [checkin_date for _, checkin_date in activity] == [
            "2026-07-16",
            "2026-07-01",
        ]

adr/0002-solid-functional-core.md

# ADR-0002: SOLID + functional-core / imperative-shell refactor
 
**Status**: Accepted
**Date**: 2026-07-19
 
## Context
 
Pass 1's `app/repository.py` mixed data access with domain-object hydration; `app/main.py`'s
route handlers called `repo.*` functions directly. Nothing was formally WRONG with this for a
single-file capstone, but it does not demonstrate: (a) depending on an abstraction rather than
a concrete database module (Dependency Inversion Principle), (b) a business-rule layer
independently testable without a database (Single Responsibility Principle), or (c) a pure
computational core separated from the parts that touch state (functional core / imperative
shell, topics 22/23).
 
## Decision
 
- Introduce `app/ports.py`: a `HabitRepository` `Protocol` (structural typing, PEP 544) naming
  every operation the application layer needs from storage.
- Introduce `app/services.py`: `HabitService`, depending only on `HabitRepository`, owning the
  business rules (e.g., "a check-in needs an owned habit") that used to live inline in route
  handlers.
- `app/repository_sqlite.py`'s `SqliteHabitRepository` is the ONE concrete adapter that ships in
  production, satisfying `HabitRepository` by shape, never by inheritance.
- `app/domain.py` splits into pure functions (`current_streak`, `longest_streak_ever`,
  `longest_streak_ever_naive`) plus a thin `Habit` shell that delegates to them -- the
  computation itself needs no database, no HTTP server, and no `Habit` instance to be tested.
 
## Consequences
 
- **Positive (Open/Closed proof)**: `tests/test_services.py` adds `InMemoryHabitRepository` --
  a SECOND class satisfying `HabitRepository` -- with zero edits to `services.py`, `ports.py`,
  or `repository_sqlite.py`. `HabitService` accepts it exactly as it accepts the real adapter.
- **Positive**: the functional core (`domain.py`'s free functions) is independently
  benchmarkable (`bench/benchmark_algorithm.py`) without spinning up a database or server.
- **Trade-off**: one more file, one more layer of indirection (`main.py` -> `HabitService` ->
  `HabitRepository` -> `SqliteHabitRepository`) than Pass 1's flatter shape -- justified here
  because the syllabus spec requires a DEMONSTRATED OCP extension point, not just a claim of one.
 
## Verification
 
Behavior preservation: `tests/test_app.py`'s full integration suite (auth, CRUD, ownership,
injection-safety) passes unchanged in spirit against the refactored code (see this page's Step
2 transcript). OCP: `tests/test_services.py::TestHabitServiceWithInMemoryRepository` passes
against `InMemoryHabitRepository`, added without touching any shipped `app/` file.

$ pytest tests/test_services.py -v
collecting ... collected 6 items
 
tests/test_services.py::TestHabitServiceWithInMemoryRepository::test_create_then_get_round_trips PASSED [ 16%]
tests/test_services.py::TestHabitServiceWithInMemoryRepository::test_checkin_updates_streak PASSED [ 33%]
tests/test_services.py::TestHabitServiceWithInMemoryRepository::test_checkin_against_a_habit_the_caller_does_not_own_is_rejected PASSED [ 50%]
tests/test_services.py::TestHabitServiceWithInMemoryRepository::test_search_filters_by_substring PASSED [ 66%]
tests/test_services.py::TestHabitServiceWithInMemoryRepository::test_archiving_hides_from_the_default_list PASSED [ 83%]
tests/test_services.py::TestHabitServiceWithInMemoryRepository::test_recent_activity_orders_newest_first PASSED [100%]
 
6 passed in 0.69s

$ curl -s -X POST http://127.0.0.1:8101/auth/register -H "Content-Type: application/json" \
  -d '{"username":"asha","password":"correct-horse-battery-staple"}'
{"id":1,"username":"asha","created_at":"2026-07-19 01:51:20"}
 
$ TOKEN=$(curl -s -X POST http://127.0.0.1:8101/auth/login -H "Content-Type: application/json" \
  -d '{"username":"asha","password":"correct-horse-battery-staple"}' | python3 -c \
  "import json,sys; print(json.load(sys.stdin)['access_token'])")
 
$ curl -s -X POST http://127.0.0.1:8101/habits -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" -d '{"name":"Read 20 pages"}'
{"id":1,"name":"Read 20 pages","created_at":"2026-07-19 01:51:21","archived":false,"checkin_count":0,"current_streak":0}
 
$ curl -s -X POST http://127.0.0.1:8101/habits/1/checkins -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" -d '{"checkin_date":"2026-07-19"}'
{"habit_id":1,"checkin_date":"2026-07-19","checkin_count":1,"current_streak":1}

app/digest.py

"""capstone-solid-core: the digest hot path -- SEQUENTIAL and CONCURRENT strategies that
compute the IDENTICAL result (topic 24 Concurrency & Parallelism). For every one of a user's
habits, computes its current streak, its longest-streak-ever (Step 3's O(n) algorithm, topic
25), and its total check-in count -- for a user with many habits and long histories, this is
genuinely CPU-bound per habit (rebuilding the domain object's hash-set from every stored
check-in row, then an O(n) scan over it), so it is the hot path this capstone makes concurrent.
 
DELIBERATELY NOT wired behind a synchronous HTTP route: spawning a fresh `ProcessPoolExecutor`
on every incoming request would pay process-spawn overhead per request, the opposite of what
Step 3's own benchmark shows (see `bench/benchmark_concurrency.py` -- overhead dominates at
small scale). This module is called directly instead, the way a real system would run it -- a
periodic batch/reporting job (a nightly digest email, an admin report) -- which is exactly
where `ProcessPoolExecutor`'s per-call spawn cost is amortized across a large, one-shot batch
rather than paid on every request. `tests/test_app.py`'s `TestDigestSequentialAndConcurrentAgree`
proves both strategies agree; `bench/benchmark_concurrency.py` measures the real speedup.
 
Each worker opens its OWN `sqlite3.Connection` -- a connection cannot cross a process boundary
(it holds an OS file handle and internal C state that does not survive `pickle`), so `digest.py`
passes only the `db_path` (a plain string) to each worker and lets it reconnect. `PRAGMA
journal_mode=WAL` (set in `repository_sqlite.get_connection`) is what makes many concurrent
READER connections to the same SQLite file safe (sqlite.org/wal.html): "WAL provides more
concurrency as readers do not block writers and a writer does not block readers."
"""
 
from __future__ import annotations
 
import sqlite3
from concurrent.futures import ProcessPoolExecutor
from dataclasses import dataclass
from datetime import date
 
from .repository_sqlite import SqliteHabitRepository
 
 
@dataclass(frozen=True, slots=True)
class HabitDigest:
    habit_id: int
    name: str
    current_streak: int
    longest_streak_ever: int
    checkin_count: int
 
 
def _digest_for_habit(
    db_path: str, user_id: int, habit_id: int, today_iso: str
) -> HabitDigest:
    """Runs INSIDE a worker (this same function is called directly by the sequential path
    below, and via `ProcessPoolExecutor` by the concurrent path -- same function, same result,
    only the CALLER differs). Opens its own connection: state that must cross a process
    boundary has to be rebuilt on the other side, never shared by reference (DD-33
    taming-state, applied to concurrency instead of just to the DB-vs-domain-object split)."""
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    try:
        repo = SqliteHabitRepository(conn)
        habit = repo.get_habit(habit_id, user_id)
        assert habit is not None  # => the caller only ever passes ids it just listed
        today = date.fromisoformat(today_iso)
        return HabitDigest(
            habit_id=habit.id,
            name=habit.name,
            current_streak=habit.current_streak(today),
            longest_streak_ever=habit.longest_streak_ever(),  # => Step 3's O(n) algorithm
            checkin_count=habit.checkin_count(),
        )
    finally:
        conn.close()
 
 
def sequential_digest(
    db_path: str, user_id: int, habit_ids: list[int], today: date
) -> list[HabitDigest]:
    """BEFORE: one habit at a time, entirely inside the request-handling process."""
    today_iso = today.isoformat()
    return [_digest_for_habit(db_path, user_id, hid, today_iso) for hid in habit_ids]
 
 
def concurrent_digest(
    db_path: str,
    user_id: int,
    habit_ids: list[int],
    today: date,
    max_workers: int = 4,
) -> list[HabitDigest]:
    """AFTER: each habit's load + O(n) longest-streak scan runs in its OWN OS process
    (`concurrent.futures.ProcessPoolExecutor`, Python standard library since 3.2 --
    docs.python.org/3/library/concurrent.futures.html) -- genuine parallelism across CPU
    cores for this CPU-bound per-habit work. A THREAD pool would stay serialized on the
    CPU-bound parts of this same code path (building the hash-set, walking it) because of the
    GIL; a PROCESS pool does not share a GIL at all."""
    today_iso = today.isoformat()
    with ProcessPoolExecutor(max_workers=max_workers) as pool:
        results = list(
            pool.map(
                _digest_for_habit,
                [db_path] * len(habit_ids),
                [user_id] * len(habit_ids),
                habit_ids,
                [today_iso] * len(habit_ids),
            )
        )
    return results

app / migration_v3.sql

-- capstone-solid-core: Step 3's EXPLAIN-guided index (topic 26 Advanced SQL & Query
-- Performance, topic 10 co-22 schema-migration). Applied once at PRAGMA user_version 2 -> 3
-- by app/repository_sqlite.py's init_db().
--
-- WHY: GET /habits/activity/recent (Step 3's new "recent activity across ALL my habits"
-- endpoint) needs each user's most recent check-ins, newest first, without caring which
-- specific habit each one belongs to. Reached through the ORIGINAL normalized schema, that
-- query has to JOIN checkins to habits just to filter by user_id, then SORT the joined rows by
-- checkin_date -- confirmed for real via EXPLAIN QUERY PLAN (see bench/explain_query_plan.sh):
-- `SEARCH h USING COVERING INDEX idx_habits_user_id (user_id=?)` then
-- `SEARCH c USING COVERING INDEX sqlite_autoindex_checkins_1 (habit_id=?)` (NOT a full table
-- scan -- both sides of the join already use an index) followed by `USE TEMP B-TREE FOR ORDER BY`.
--
-- FIX: a deliberate, DOCUMENTED denormalization (topic 26 co-XX denormalization-tradeoffs) --
-- copy `user_id` onto `checkins` (it never changes after a habit is created, so there is no
-- update-anomaly risk this specific denormalization introduces) and index (user_id,
-- checkin_date DESC). The query becomes a single ordered index walk with NO join and NO sort:
-- `SEARCH checkins USING INDEX idx_checkins_user_id_date (user_id=?)`.
--
-- Backfill uses SQLite's `UPDATE ... FROM` (correlated update), supported since SQLite 3.33.0
-- (2020-08-14, https://sqlite.org/releaselog/3_33_0.html) -- well below this app's pinned
-- runtime (Python 3.13's bundled SQLite is materially newer). The new column stays nullable at
-- the schema level (SQLite cannot add a NOT NULL column with no default to a non-empty table
-- in one step); repository_sqlite.py always supplies it going forward, so it is never actually
-- null in practice.
ALTER TABLE checkins
ADD COLUMN user_id INTEGER REFERENCES users (id) ON DELETE CASCADE;
 
UPDATE checkins
SET
  user_id = habits.user_id
FROM
  habits
WHERE
  habits.id = checkins.habit_id;
 
CREATE INDEX IF NOT EXISTS idx_checkins_user_id_date ON checkins (user_id, checkin_date DESC);

bench/explain_query_plan.sh

#!/usr/bin/env bash
# capstone-solid-core: Step 3's EXPLAIN-guided-index demonstration (topic 26 Advanced SQL &
# Query Performance). This app's database is SQLite, not PostgreSQL -- topic 26's own teaching
# engine is PostgreSQL specifically for `EXPLAIN ANALYZE`'s execution-time statistics
# (sqlite.org/lang_explain.html: SQLite's grammar has EXPLAIN and EXPLAIN QUERY PLAN, no
# ANALYZE keyword -- confirmed by reading that page directly). The TECHNIQUE topic 26 teaches
# -- read what the planner will do, add an index that removes an expensive step, confirm with
# the planner AND a real timing measurement -- is identical here, applied to the engine this
# specific app actually runs (sqlite.org/eqp.html documents EXPLAIN QUERY PLAN's own output).
# The timed before/after comparison lives in bench/benchmark_sql_tuning.py (perf_counter has
# far more resolution than this CLI's own `.timer` for a query this fast); this script only
# shows the real, captured QUERY PLAN shape changing.
set -euo pipefail
 
DB="/tmp/capstone-solid-core-eqp-bench.db"
rm -f "$DB"
 
echo "==> seeding: 1 user, 3 habits, 200,001 total check-ins (normalized schema, no denormalized column yet)"
sqlite3 "$DB" <<'SQL'
CREATE TABLE users (id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT NOT NULL UNIQUE, password_hash TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT (datetime('now')));
CREATE TABLE habits (id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, name TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT (datetime('now')), archived INTEGER NOT NULL DEFAULT 0);
CREATE TABLE checkins (id INTEGER PRIMARY KEY AUTOINCREMENT, habit_id INTEGER NOT NULL REFERENCES habits(id) ON DELETE CASCADE, checkin_date TEXT NOT NULL, UNIQUE(habit_id, checkin_date));
CREATE INDEX idx_habits_user_id ON habits(user_id);
CREATE INDEX idx_checkins_habit_id ON checkins(habit_id);
INSERT INTO users (username, password_hash) VALUES ('bench_user', 'unused');
INSERT INTO habits (user_id, name) VALUES (1, 'Habit A'), (1, 'Habit B'), (1, 'Habit C');
 
WITH RECURSIVE seq(n) AS (SELECT 0 UNION ALL SELECT n + 1 FROM seq WHERE n < 66666)
INSERT INTO checkins (habit_id, checkin_date) SELECT 1, date('1990-01-01', n || ' days') FROM seq;
WITH RECURSIVE seq(n) AS (SELECT 0 UNION ALL SELECT n + 1 FROM seq WHERE n < 66666)
INSERT INTO checkins (habit_id, checkin_date) SELECT 2, date('2050-01-01', n || ' days') FROM seq;
WITH RECURSIVE seq(n) AS (SELECT 0 UNION ALL SELECT n + 1 FROM seq WHERE n < 66666)
INSERT INTO checkins (habit_id, checkin_date) SELECT 3, date('2110-01-01', n || ' days') FROM seq;
SQL
echo "==> seeded: $(sqlite3 "$DB" 'SELECT COUNT(*) FROM checkins') rows in checkins"
 
echo
echo "=== BEFORE: EXPLAIN QUERY PLAN for the recent-activity query (JOIN, no denormalized user_id) ==="
sqlite3 "$DB" "EXPLAIN QUERY PLAN SELECT h.id, c.checkin_date FROM checkins c JOIN habits h ON h.id = c.habit_id WHERE h.user_id = 1 ORDER BY c.checkin_date DESC LIMIT 20;"
 
echo
echo "==> applying migration_v3.sql (denormalize checkins.user_id + composite index)"
sqlite3 "$DB" <"$(dirname "${BASH_SOURCE[0]}")/../app/migration_v3.sql"
 
echo
echo "=== AFTER: EXPLAIN QUERY PLAN for the recent-activity query (single index, no join, no sort) ==="
sqlite3 "$DB" "EXPLAIN QUERY PLAN SELECT habit_id, checkin_date FROM checkins WHERE user_id = 1 ORDER BY checkin_date DESC LIMIT 20;"
 
echo
echo "=== correctness: BEFORE and AFTER return the identical 20 rows ==="
diff \
	<(sqlite3 "$DB" "SELECT h.id, c.checkin_date FROM checkins c JOIN habits h ON h.id = c.habit_id WHERE h.user_id = 1 ORDER BY c.checkin_date DESC LIMIT 20;") \
	<(sqlite3 "$DB" "SELECT habit_id, checkin_date FROM checkins WHERE user_id = 1 ORDER BY checkin_date DESC LIMIT 20;") &&
	echo "IDENTICAL -- the index changed the PLAN, not the RESULT"

$ pytest tests/test_domain.py::TestLongestStreakEver "tests/test_app.py::TestRecentActivity" \
  "tests/test_app.py::TestDigestSequentialAndConcurrentAgree" -v
collecting ... collected 29 items
 
tests/test_domain.py::TestLongestStreakEver::test_empty_history_has_longest_streak_zero PASSED
tests/test_domain.py::TestLongestStreakEver::test_single_checkin_is_a_streak_of_one PASSED
tests/test_domain.py::TestLongestStreakEver::test_two_separate_single_day_checkins_do_not_combine PASSED
tests/test_domain.py::TestLongestStreakEver::test_a_past_streak_can_be_longer_than_the_current_one PASSED
tests/test_domain.py::TestLongestStreakEver::test_overlapping_runs_pick_the_longest_one PASSED
tests/test_domain.py::TestLongestStreakEver::test_algorithms_agree_on_random_histories[0..19] PASSED (20 seeds)
tests/test_app.py::TestRecentActivity::test_recent_activity_spans_every_habit_newest_first PASSED
tests/test_app.py::TestRecentActivity::test_recent_activity_is_scoped_to_the_caller PASSED
tests/test_app.py::TestRecentActivity::test_recent_activity_requires_auth PASSED
tests/test_app.py::TestDigestSequentialAndConcurrentAgree::test_sequential_and_concurrent_digest_return_identical_results PASSED
 
29 passed in 1.05s

$ bash bench/explain_query_plan.sh
==> seeding: 1 user, 3 habits, 200,001 total check-ins (normalized schema, no denormalized column yet)
==> seeded: 200001 rows in checkins
 
=== BEFORE: EXPLAIN QUERY PLAN for the recent-activity query (JOIN, no denormalized user_id) ===
QUERY PLAN
|--SEARCH h USING COVERING INDEX idx_habits_user_id (user_id=?)
|--SEARCH c USING COVERING INDEX sqlite_autoindex_checkins_1 (habit_id=?)
`--USE TEMP B-TREE FOR ORDER BY
 
==> applying migration_v3.sql (denormalize checkins.user_id + composite index)
 
=== AFTER: EXPLAIN QUERY PLAN for the recent-activity query (single index, no join, no sort) ===
QUERY PLAN
`--SEARCH checkins USING INDEX idx_checkins_user_id_date (user_id=?)
 
=== correctness: BEFORE and AFTER return the identical 20 rows ===
IDENTICAL -- the index changed the PLAN, not the RESULT

bench/benchmark_algorithm.py

"""capstone-solid-core: Step 3's algorithm benchmark (topic 25 Advanced Algorithms). Measures
`longest_streak_ever_naive` (O(n log n), sort-based) against `longest_streak_ever` (O(n),
hash-set-based) over the SAME randomly generated check-in history, growing `n`, using
`time.perf_counter()` -- the standard library's monotonic, highest-resolution timer
(docs.python.org/3/library/time.html#time.perf_counter): "always returns a monotonic value ...
should be used for measuring performance." Every number printed below comes from actually
running both functions against the same generated data; this script prints nothing it did not
just compute.
 
Run: python3 -m bench.benchmark_algorithm   (from capstone-solid-core/code/, inside the venv)
"""
 
from __future__ import annotations
 
import random
import time
from datetime import date, timedelta
 
from app.domain import longest_streak_ever, longest_streak_ever_naive
 
 
def _generate_checkin_history(n: int, seed: int) -> set[date]:
    """A synthetic history of `n` DISTINCT calendar days, scattered (not necessarily
    consecutive) over a window wide enough to hold them -- exercises the general case, not
    just one long unbroken streak."""
    rng = random.Random(seed)
    base = date(1990, 1, 1)
    window = (
        n * 3
    )  # => scattered across a window 3x wider than n, so most are NOT adjacent
    offsets = rng.sample(range(window), n)
    return {base + timedelta(days=offset) for offset in offsets}
 
 
def _time_once(func, checkin_dates: set[date]) -> tuple[int, float]:
    start = time.perf_counter()
    result = func(checkin_dates)
    elapsed = time.perf_counter() - start
    return result, elapsed
 
 
def main() -> None:
    sizes = [1_000, 10_000, 100_000, 500_000]
    print(
        f"{'n':>10}  {'naive O(n log n) (s)':>22}  {'optimized O(n) (s)':>20}  {'speedup':>9}"
    )
    for n in sizes:
        checkin_dates = _generate_checkin_history(n, seed=42)
 
        naive_result, naive_elapsed = _time_once(
            longest_streak_ever_naive, checkin_dates
        )
        optimized_result, optimized_elapsed = _time_once(
            longest_streak_ever, checkin_dates
        )
 
        assert naive_result == optimized_result, (
            f"MISMATCH at n={n}: naive={naive_result} optimized={optimized_result} "
            "-- the benchmark is invalid if the two algorithms disagree"
        )
 
        speedup = (
            naive_elapsed / optimized_elapsed if optimized_elapsed > 0 else float("inf")
        )
        print(
            f"{n:>10}  {naive_elapsed:>22.6f}  {optimized_elapsed:>20.6f}  {speedup:>8.2f}x"
        )
 
 
if __name__ == "__main__":
    main()

$ python3 -m bench.benchmark_algorithm
         n    naive O(n log n) (s)    optimized O(n) (s)    speedup
      1000                0.000348              0.000134      2.59x
     10000                0.003821              0.001343      2.85x
    100000                0.044984              0.016978      2.65x
    500000                0.297311              0.103849      2.86x

bench/benchmark_concurrency.py

"""capstone-solid-core: Step 3's concurrency benchmark (topic 24 Concurrency & Parallelism).
Seeds a real SQLite database with many habits, each with a long check-in history, then times
`sequential_digest` against `concurrent_digest` (app/digest.py) over the SAME database with
`time.perf_counter()`. Both are asserted to return the identical set of results BEFORE the
timing numbers are printed -- a benchmark that skipped the correctness check could silently
compare a slow-but-right implementation to a fast-but-wrong one.
 
Run (large, realistic scale -- the default): python3 -m bench.benchmark_concurrency
Run (small scale, to reproduce ADR-0003's small-workload finding):
    python3 -m bench.benchmark_concurrency --num-habits 16 --checkins-per-habit 8000
(from capstone-solid-core/code/, inside the venv)
"""
 
from __future__ import annotations
 
import argparse
import os
import random
import sys
import time
from datetime import date, timedelta
 
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
from app.digest import concurrent_digest, sequential_digest
from app.models import HabitCreate
from app.repository_sqlite import SqliteHabitRepository, get_connection, init_db
 
 
def _seed_database(
    db_path: str, num_habits: int, checkins_per_habit: int, seed: int
) -> list[int]:
    """Creates one user with `num_habits` habits, each with `checkins_per_habit` scattered,
    non-consecutive check-ins -- large enough per-habit histories that
    `longest_streak_ever`'s O(n) scan is genuinely non-trivial work per habit.
 
    Uses `executemany` + ONE commit at the end -- not `SqliteHabitRepository.record_checkin`
    (which commits per call, correct for a live request but far too slow for seeding hundreds
    of thousands of synthetic rows). This is seed-data generation, not the code path under
    benchmark; correctness of the PRODUCTION per-request path is already covered by
    tests/test_app.py's integration tests."""
    rng = random.Random(seed)
    init_db(db_path)
    conn = get_connection(db_path)
    from app.repository_sqlite import create_user
 
    user = create_user(conn, "bench_user", "unused-hash-for-benchmark-only")
    repo = SqliteHabitRepository(conn)
    habit_ids: list[int] = []
    base = date(1990, 1, 1)
    for i in range(num_habits):
        habit = repo.create_habit(user.id, HabitCreate(name=f"Habit {i}"))
        habit_ids.append(habit.id)
        window = checkins_per_habit * 3
        offsets = rng.sample(range(window), checkins_per_habit)
        rows = [
            (habit.id, user.id, (base + timedelta(days=offset)).isoformat())
            for offset in offsets
        ]
        conn.executemany(
            "INSERT INTO checkins (habit_id, user_id, checkin_date) VALUES (?, ?, ?)",
            rows,
        )
    conn.commit()
    conn.close()
    return habit_ids
 
 
def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--num-habits",
        type=int,
        default=40,
        help="Number of habits to seed (default: 40, the large-scale scenario).",
    )
    parser.add_argument(
        "--checkins-per-habit",
        type=int,
        default=25_000,
        help="Check-ins per habit to seed (default: 25000, the large-scale scenario).",
    )
    args = parser.parse_args()
 
    db_path = "/tmp/capstone-solid-core-concurrency-bench.db"
    if os.path.exists(db_path):
        os.remove(db_path)
    num_habits = args.num_habits
    checkins_per_habit = args.checkins_per_habit
    print(
        f"Seeding {num_habits} habits x {checkins_per_habit} check-ins each "
        f"({num_habits * checkins_per_habit} total rows)..."
    )
    habit_ids = _seed_database(db_path, num_habits, checkins_per_habit, seed=7)
    today = date(1990, 1, 1) + timedelta(days=checkins_per_habit * 3)
 
    start = time.perf_counter()
    sequential_result = sequential_digest(
        db_path, user_id=1, habit_ids=habit_ids, today=today
    )
    sequential_elapsed = time.perf_counter() - start
 
    start = time.perf_counter()
    concurrent_result = concurrent_digest(
        db_path, user_id=1, habit_ids=habit_ids, today=today, max_workers=4
    )
    concurrent_elapsed = time.perf_counter() - start
 
    assert sorted(sequential_result, key=lambda d: d.habit_id) == sorted(
        concurrent_result, key=lambda d: d.habit_id
    ), "MISMATCH -- sequential and concurrent digests must return identical results"
 
    speedup = (
        sequential_elapsed / concurrent_elapsed
        if concurrent_elapsed > 0
        else float("inf")
    )
    print(f"sequential_digest:  {sequential_elapsed:.4f}s")
    print(f"concurrent_digest:  {concurrent_elapsed:.4f}s  (max_workers=4)")
    print(f"speedup:            {speedup:.2f}x")
    print("results match:      True (asserted above before these numbers were printed)")
 
 
if __name__ == "__main__":
    main()

$ python3 -m bench.benchmark_concurrency
Seeding 40 habits x 25000 check-ins each (1000000 total rows)...
sequential_digest:  0.5843s
concurrent_digest:  0.3029s  (max_workers=4)
speedup:            1.93x
results match:      True (asserted above before these numbers were printed)

bench/benchmark_sql_tuning.py

"""capstone-solid-core: Step 3's SQL-tuning wall-clock benchmark (topic 26 Advanced SQL &
Query Performance), pairing bench/explain_query_plan.sh's real EXPLAIN QUERY PLAN output with
a PRECISE timing comparison -- the `sqlite3` CLI's own `.timer` rounds "real" time to 3 decimal
places, too coarse for a query this fast; `time.perf_counter()` (the standard library's
monotonic, highest-resolution timer) does not have that limitation.
 
Seeds the SAME shape of data as explain_query_plan.sh (1 user, 3 habits, 200,001 total
check-ins) directly through the Python `sqlite3` module, runs the recent-activity query 200
times BEFORE the index and 200 times AFTER, and reports the total elapsed time for each batch.
Every number below comes from an actual run against a real SQLite file; nothing here is
estimated or assumed.
 
Run: python3 -m bench.benchmark_sql_tuning   (from capstone-solid-core/code/, inside the venv)
"""
 
from __future__ import annotations
 
import os
import sqlite3
import time
from datetime import date, timedelta
from pathlib import Path
 
REPETITIONS = 200
 
 
def _seed(db_path: str) -> None:
    conn = sqlite3.connect(db_path)
    conn.executescript(
        """
        CREATE TABLE users (id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT NOT NULL UNIQUE,
            password_hash TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT (datetime('now')));
        CREATE TABLE habits (id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
            name TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT (datetime('now')),
            archived INTEGER NOT NULL DEFAULT 0);
        CREATE TABLE checkins (id INTEGER PRIMARY KEY AUTOINCREMENT,
            habit_id INTEGER NOT NULL REFERENCES habits(id) ON DELETE CASCADE,
            checkin_date TEXT NOT NULL, UNIQUE(habit_id, checkin_date));
        CREATE INDEX idx_habits_user_id ON habits(user_id);
        CREATE INDEX idx_checkins_habit_id ON checkins(habit_id);
        INSERT INTO users (username, password_hash) VALUES ('bench_user', 'unused');
        INSERT INTO habits (user_id, name) VALUES (1, 'Habit A'), (1, 'Habit B'), (1, 'Habit C');
        """
    )
    base = date(1990, 1, 1)
    for habit_id, offset_start in ((1, 0), (2, 22_000), (3, 44_000)):
        rows = [
            (habit_id, (base + timedelta(days=offset_start + i)).isoformat())
            for i in range(66_667)
        ]
        conn.executemany(
            "INSERT INTO checkins (habit_id, checkin_date) VALUES (?, ?)", rows
        )
    conn.commit()
    conn.close()
 
 
def _time_query(
    conn: sqlite3.Connection, sql: str, params: tuple[object, ...]
) -> float:
    start = time.perf_counter()
    for _ in range(REPETITIONS):
        conn.execute(sql, params).fetchall()
    return time.perf_counter() - start
 
 
def main() -> None:
    db_path = "/tmp/capstone-solid-core-sql-tuning-bench.db"
    if os.path.exists(db_path):
        os.remove(db_path)
    print(f"Seeding {db_path} with 200,001 check-ins across 3 habits, 1 user...")
    _seed(db_path)
 
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
 
    before_sql = (
        "SELECT h.id, c.checkin_date FROM checkins c "
        "JOIN habits h ON h.id = c.habit_id "
        "WHERE h.user_id = ? ORDER BY c.checkin_date DESC LIMIT ?"
    )
    before_result = conn.execute(before_sql, (1, 20)).fetchall()
    before_elapsed = _time_query(conn, before_sql, (1, 20))
 
    migration_v3 = (
        Path(__file__).parent.parent / "app" / "migration_v3.sql"
    ).read_text(encoding="utf-8")
    conn.executescript(migration_v3)
 
    after_sql = (
        "SELECT habit_id, checkin_date FROM checkins "
        "WHERE user_id = ? ORDER BY checkin_date DESC LIMIT ?"
    )
    after_result = conn.execute(after_sql, (1, 20)).fetchall()
    after_elapsed = _time_query(conn, after_sql, (1, 20))
 
    before_rows = [(r["id"], r["checkin_date"]) for r in before_result]
    after_rows = [(r["habit_id"], r["checkin_date"]) for r in after_result]
    assert before_rows == after_rows, "MISMATCH -- the index must not change the RESULT"
 
    speedup = before_elapsed / after_elapsed if after_elapsed > 0 else float("inf")
    print(
        f"\n{REPETITIONS} repetitions of the recent-activity query, same 200,001-row DB:"
    )
    print(
        f"  BEFORE (join + temp b-tree sort): {before_elapsed:.4f}s total, "
        f"{before_elapsed / REPETITIONS * 1000:.4f}ms/query"
    )
    print(
        f"  AFTER  (single ordered index scan): {after_elapsed:.4f}s total, "
        f"{after_elapsed / REPETITIONS * 1000:.4f}ms/query"
    )
    print(f"  speedup: {speedup:.2f}x")
    print(
        "  results identical (asserted above before these numbers were printed): True"
    )
 
    conn.close()
 
 
if __name__ == "__main__":
    main()

$ python3 -m bench.benchmark_sql_tuning
Seeding /tmp/capstone-solid-core-sql-tuning-bench.db with 200,001 check-ins across 3 habits, 1 user...
 
200 repetitions of the recent-activity query, same 200,001-row DB:
  BEFORE (join + temp b-tree sort): 0.0055s total, 0.0276ms/query
  AFTER  (single ordered index scan): 0.0033s total, 0.0167ms/query
  speedup: 1.65x
  results identical (asserted above before these numbers were printed): True

adr/0003-concurrency-and-indexing.md

# ADR-0003: Concurrent digest, an O(n) algorithm, and an EXPLAIN-guided index
 
**Status**: Accepted
**Date**: 2026-07-19
 
## Context
 
Step 3 needed a genuine hot path to parallelize (topic 24), a genuine algorithmic improvement
(topic 25), and a genuine index tuned by reading a real query plan (topic 26). All three
numbers on this page are MEASURED, not assumed -- two of the three measurements below
contradicted the naive expectation on the first attempt, which is exactly why they were
measured instead of asserted.
 
## Decision 1: `longest_streak_ever` -- O(n) over integer ordinals, not `date`/`timedelta`
 
A first O(n) implementation, written directly against `date`/`timedelta` objects, was
benchmarked (`bench/benchmark_algorithm.py`) against the O(n log n) `sorted()`-based baseline
and was SLOWER at every size up to n=500,000 -- Python's built-in `sorted()` runs in optimized C
with low per-comparison overhead, while constructing a fresh `timedelta` object on every loop
iteration in pure Python is comparatively expensive. Converting each `date` to its integer
`toordinal()` once, then running the identical algorithm over `set[int]` with plain `+1`/`-1`
arithmetic, measured 2.59x-2.86x FASTER than the baseline across the same range. **Decision**:
ship the ordinal-based version. Big-O describes asymptotic growth; it does not by itself
guarantee a win at any one concrete n in a language with non-trivial per-object overhead --
that has to be measured, and here it changed which implementation actually shipped.
 
## Decision 2: digest concurrency -- `ProcessPoolExecutor`, sized for real, not toy, workloads
 
`app/digest.py` computes each habit's streak data independently -- CPU-bound Python work
(rebuilding a hash-set from every stored check-in row, then the O(n) scan), which a THREAD pool
would not meaningfully parallelize (GIL-bound), so a `ProcessPoolExecutor` was chosen. It is
called directly as a batch job (not behind a synchronous HTTP route -- see digest.py's own
docstring for why). First benchmarked at a modest scale (16 habits x 8,000 check-ins --
reproducible via `python3 -m bench.benchmark_concurrency --num-habits 16
--checkins-per-habit 8000`), `concurrent_digest` was measurably SLOWER than
`sequential_digest` on every repeated run -- spawning worker processes and importing
`pydantic` fresh in each one costs tens of milliseconds per process, which dominated a
workload whose real per-habit work was only a few milliseconds. The exact ratio is itself
noisy at this small a scale (eight repeated runs on the machine that authored this ADR
measured between 0.13x and 0.69x -- roughly 1.4x to 8x slower than sequential, never
faster) because a fixed, OS-scheduled process-spawn cost competes with whatever else the
machine is doing at that instant; the DIRECTION (slower) reproduces every run, so no single
precise ratio is quoted here. At a larger, still realistic scale (40 habits x 25,000
check-ins = 1,000,000 rows), `concurrent_digest` measured 1.7x-2.3x FASTER across repeated
runs -- the fixed process-pool startup cost is paid ONCE per `with
ProcessPoolExecutor(...)` block, and is only worth paying
when the total real work clears that fixed cost by a wide margin. **Decision**: ship
`concurrent_digest`, and document the small-workload finding here rather than hide it -- it is
the actual engineering lesson topic 24 teaches: concurrency has overhead, and the right call
depends on the workload's real size, not on "concurrent is always faster."
 
## Decision 3: denormalize `checkins.user_id`, guided by real `EXPLAIN QUERY PLAN` output
 
This app's database is SQLite, not PostgreSQL -- topic 26's own teaching engine is PostgreSQL
specifically for `EXPLAIN ANALYZE`'s execution-time statistics (confirmed by reading
sqlite.org/lang_explain.html directly: SQLite's grammar has `EXPLAIN` and
`EXPLAIN QUERY PLAN`, no `ANALYZE` keyword). Reading this app's OWN engine's real
`EXPLAIN QUERY PLAN` output for a "recent activity across all my habits" query showed TWO
indexed searches -- `SEARCH h USING COVERING INDEX idx_habits_user_id (user_id=?)` then
`SEARCH c USING COVERING INDEX sqlite_autoindex_checkins_1 (habit_id=?)` -- plus
`USE TEMP B-TREE FOR ORDER BY` (NOT a full table scan; see `bench/explain_query_plan.sh`):
the normalized schema still forces a join from `checkins` to `habits` just to filter by
`user_id`, then a sort, even with both sides of the join reached through an index.
**Decision**: denormalize `user_id` onto `checkins` (it never changes after a habit is
created, so this specific denormalization introduces no update-anomaly risk) and add a
composite index `(user_id, checkin_date DESC)` (`migration_v3.sql`). The SAME query re-read via
`EXPLAIN QUERY PLAN` afterward shows a single `SEARCH checkins USING INDEX
idx_checkins_user_id_date (user_id=?)` -- no join, no sort.
 
## Consequences
 
- **Positive**: all three changes are measured, not assumed, and two of the three measurements
  overturned the naive first guess -- documented here rather than silently discarded.
- **Trade-off**: `checkins.user_id` duplicates data already reachable via `habit_id ->
habits.user_id` -- an accepted, documented redundancy (topic 26 denormalization trade-off),
  not an oversight; `repository_sqlite.SqliteHabitRepository.record_checkin` is the ONE place
  this app ever writes a `checkins` row, so the copy cannot drift.
 
## Verification
 
`tests/test_domain.py::TestLongestStreakEver` cross-checks the O(n) and O(n log n)
implementations agree on 20 randomized histories before either is trusted. `tests/test_app.py`'s
`TestDigestSequentialAndConcurrentAgree` asserts `sequential_digest` and `concurrent_digest`
return identical results. `bench/explain_query_plan.sh` + `bench/benchmark_sql_tuning.py`
confirm the before/after rows are identical (`diff` + an `assert`) -- the index changed the
PLAN, never the RESULT.

ci.yml

# capstone-solid-core: Step 4's lint -> test -> build pipeline (topic 30 co-08/co-09/co-10,
# reusing the same three-stage shape this plan's own software-engineering-practices capstone
# already validated). Copy this file to `.github/workflows/ci.yml` in a real clone of `code/`
# to run it on GitHub Actions. Validated with `actionlint` -- zero findings.
name: ci
on:
  pull_request:
 
jobs:
  lint: # => STAGE 1 -- the SAME ruff commands scripts/run_ci_locally.sh already runs locally
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v7
      - uses: actions/setup-python@v6
        with:
          python-version: "3.13"
      - run: pip install -r requirements.txt
      - run: ruff check . # => a lint FAILURE stops the pipeline here -- test and build never start
      - run: ruff format --check .
 
  test: # => STAGE 2 -- runs ONLY if lint passed
    needs: lint # => the GATE -- no lint pass, no test run
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v7
      - uses: actions/setup-python@v6
        with:
          python-version: "3.13"
      - run: pip install -r requirements.txt
      - env:
          CAPSTONE_SOLID_CORE_DB_PATH: /tmp/ci-habits.db
          CAPSTONE_SOLID_CORE_AUTH_SECRET: ci-only-not-a-real-secret
        run: pytest -q # => a test FAILURE stops the pipeline here -- build never starts
 
  build: # => STAGE 3 -- runs ONLY if lint AND test both passed
    needs: test # => the GATE -- no test pass, no build run
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v7
      - uses: actions/setup-python@v6
        with:
          python-version: "3.13"
      - run: pip install -r requirements.txt
      - run: python -m compileall app # => a minimal "build" -- confirms every module compiles

scripts/run_ci_locally.sh

#!/usr/bin/env bash
# capstone-solid-core: Step 4's local CI-gate runner (topic 30 co-08/co-09/co-10). Runs the
# EXACT SAME three stages ci.yml declares, in the same order, so a contributor sees the
# identical gate CI would enforce BEFORE ever opening a PR. Each stage only runs if the
# previous one exited 0 (`set -e` below is the local equivalent of ci.yml's `needs:`).
set -euo pipefail
 
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
cd "$SCRIPT_DIR/.."
 
echo "=== STAGE 1/3: lint (ruff check + ruff format --check) ==="
.venv/bin/ruff check .
.venv/bin/ruff format --check .
echo "lint: PASS"
echo
 
echo "=== STAGE 2/3: test (pytest) ==="
export CAPSTONE_SOLID_CORE_DB_PATH="/tmp/ci-local-habits.db"
export CAPSTONE_SOLID_CORE_AUTH_SECRET="ci-local-not-a-real-secret"
rm -f "$CAPSTONE_SOLID_CORE_DB_PATH"
.venv/bin/python -m pytest -q
echo "test: PASS"
echo
 
echo "=== STAGE 3/3: build (python -m compileall app) ==="
.venv/bin/python -m compileall app
echo "build: PASS"
echo
 
echo "ALL STAGES GREEN"

$ bash scripts/run_ci_locally.sh
=== STAGE 1/3: lint (ruff check + ruff format --check) ===
All checks passed!
17 files already formatted
lint: PASS
 
=== STAGE 2/3: test (pytest) ===
...............................................................          [100%]
63 passed in 1.42s
test: PASS
 
=== STAGE 3/3: build (python -m compileall app) ===
Listing 'app'...
build: PASS
 
ALL STAGES GREEN

$ # NAME/DESCRIPTION/ELAPSED/RESULT below is a mocked GitHub Actions summary (no live Actions
$ # run triggered for this page); the ruff/pytest output beneath it is real, captured output.
NAME              DESCRIPTION                          ELAPSED  RESULT
ci / lint         ruff check . && ruff format --check   1s      pass
ci / test         pytest -q                             1s      fail
ci / build        python -m compileall app              -       skipped (needs: test)
 
$ bash scripts/run_ci_locally.sh
=== STAGE 1/3: lint (ruff check + ruff format --check) ===
All checks passed!
17 files already formatted
lint: PASS
 
=== STAGE 2/3: test (pytest) ===
......................F........F...............................          [100%]
=================================== FAILURES ===================================
_ TestDigestSequentialAndConcurrentAgree.test_sequential_and_concurrent_digest_return_identical_results _
    assert len(sequential_result) == 4
>   assert all(d.current_streak == 5 for d in sequential_result)
E   assert False
 
__________ TestCurrentStreak.test_full_month_of_consecutive_checkins ___________
    for offset in range(30):
        habit.record_checkin(today - timedelta(days=offset))
>   assert habit.current_streak(today) == 30
E   AssertionError: assert 15 == 30
E    +    where current_streak = Habit(...).current_streak
 
=========================== short test summary info ============================
FAILED tests/test_app.py::TestDigestSequentialAndConcurrentAgree::test_sequential_and_concurrent_digest_return_identical_results
FAILED tests/test_domain.py::TestCurrentStreak::test_full_month_of_consecutive_checkins
2 failed, 61 passed in 1.47s
 
$ echo "exit code: $?"
exit code: 1
# => STAGE 3/3 (build) never ran -- `set -e` stopped the script the instant `pytest` exited
# => non-zero. The SAME `needs: test` gate in ci.yml stops `build` there too on GitHub Actions.
 
$ # ... app/domain.py restored from the pre-edit backup, diff confirmed empty, re-run:
$ bash scripts/run_ci_locally.sh
=== STAGE 1/3: lint (ruff check + ruff format --check) ===
All checks passed!
17 files already formatted
lint: PASS
 
=== STAGE 2/3: test (pytest) ===
...............................................................          [100%]
63 passed in 1.46s
test: PASS
 
=== STAGE 3/3: build (python -m compileall app) ===
Listing 'app'...
build: PASS
 
ALL STAGES GREEN

scripts/build_commit_history_demo.sh

#!/usr/bin/env bash
# capstone-solid-core: Step 4's commit-history demo (topic 09/30 co-01..co-04). Walks this
# capstone's OWN four ordered steps as a clean Conventional-Commits history in a throwaway
# scratch repository (mktemp -d, never nested inside this content tree -- avoids an
# embedded-git-repo hazard). Each commit below copies REAL, VERBATIM excerpts from this
# capstone's shipped app/domain.py -- not invented code -- scaled down to sizes that stay fast
# and self-contained (pure stdlib + pytest, zero network installs) so EVERY commit's suite is
# genuinely run and genuinely green here, not merely asserted. Requires this capstone's own
# `.venv` (run `bash setup.sh` once from `code/` first if it does not exist yet).
set -euo pipefail
 
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
CODE_DIR="$(cd "$SCRIPT_DIR/.." && pwd)"
if [ ! -x "$CODE_DIR/.venv/bin/pytest" ]; then
	echo "error: $CODE_DIR/.venv not found -- run 'bash setup.sh' from code/ first" >&2
	exit 1
fi
PYTEST="$CODE_DIR/.venv/bin/pytest"
RUFF="$CODE_DIR/.venv/bin/ruff"
 
WORKDIR=$(mktemp -d)
cd "$WORKDIR"
export GIT_AUTHOR_NAME="Dev" GIT_AUTHOR_EMAIL="dev@example.com"
export GIT_COMMITTER_NAME="Dev" GIT_COMMITTER_EMAIL="dev@example.com"
git -c init.defaultBranch=main init -q
git commit --allow-empty -q -m "chore: scaffold scratch repository"
git tag v1.0.0
BASE=$(git rev-parse HEAD)
 
echo "=== COMMIT 1/4: Step 1 -- import the Pass-1 baseline under a green test ==="
cat >streak.py <<'PY' # => VERBATIM: the O(n log n) sort-based algorithm Pass 1 shipped
from datetime import timedelta
 
 
def longest_streak_ever(checkin_dates):
    if not checkin_dates:
        return 0
    ordered = sorted(checkin_dates)
    longest = 1
    current = 1
    for previous_day, this_day in zip(ordered, ordered[1:]):
        if this_day - previous_day == timedelta(days=1):
            current += 1
            longest = max(longest, current)
        else:
            current = 1
    return longest
PY
cat >test_streak.py <<'PY'
from datetime import date
 
from streak import longest_streak_ever
 
 
def test_empty_history_has_no_streak():
    assert longest_streak_ever(set()) == 0
 
 
def test_three_consecutive_days_is_a_streak_of_three():
    d = date(2026, 1, 1)
    dates = {d, d.replace(day=2), d.replace(day=3)}
    assert longest_streak_ever(dates) == 3
PY
git add streak.py test_streak.py
git commit -q -m "feat(capstone): import the pass-1 longest-streak algorithm under a green test"
"$PYTEST" -q
echo "commit 1: green"
 
echo
echo "=== COMMIT 2/4: Step 2 -- SOLID/DIP: a repository Protocol + a fake, zero inheritance ==="
cat >ports.py <<'PY' # => distilled from the real app/ports.py's HabitRepository Protocol
from typing import Protocol
 
 
class HabitRepository(Protocol):
    def record_checkin(self, habit_id: int, checkin_date_iso: str) -> None: ...
    def checkin_dates(self, habit_id: int) -> set:  # noqa: UP006 -- kept minimal for the demo
        ...
PY
cat >test_ocp.py <<'PY' # => distilled from the real tests/test_services.py OCP proof
from datetime import date
 
from ports import HabitRepository
from streak import longest_streak_ever
 
 
class InMemoryHabitRepository:  # => satisfies HabitRepository by SHAPE alone, zero inheritance
    def __init__(self):
        self._by_habit = {}
 
    def record_checkin(self, habit_id: int, checkin_date_iso: str) -> None:
        self._by_habit.setdefault(habit_id, set()).add(date.fromisoformat(checkin_date_iso))
 
    def checkin_dates(self, habit_id: int) -> set:
        return self._by_habit.get(habit_id, set())
 
 
def _uses_any_habit_repository(repo: HabitRepository, habit_id: int) -> int:
    """Depends ONLY on the Protocol (topic 21 DIP) -- never on a concrete class."""
    return longest_streak_ever(repo.checkin_dates(habit_id))
 
 
def test_a_brand_new_repository_needs_zero_edits_to_existing_code():
    repo = InMemoryHabitRepository()
    repo.record_checkin(1, "2026-01-01")
    repo.record_checkin(1, "2026-01-02")
    assert _uses_any_habit_repository(repo, 1) == 2
PY
git add ports.py test_ocp.py
git commit -q -m "refactor(capstone): introduce a HabitRepository DIP port and prove OCP with a fake"
"$PYTEST" -q
echo "commit 2: green"
 
echo
echo "=== COMMIT 3/4: Step 3 -- O(n) algorithm, measured against the Step 1 baseline ==="
cat >streak_fast.py <<'PY' # => VERBATIM: the ordinal-based O(n) rewrite from app/domain.py
def longest_streak_ever_fast(checkin_dates):
    if not checkin_dates:
        return 0
    ordinals = {day.toordinal() for day in checkin_dates}
    longest = 0
    for ordinal in ordinals:
        if (ordinal - 1) in ordinals:
            continue
        run_length = 1
        probe = ordinal + 1
        while probe in ordinals:
            run_length += 1
            probe += 1
        longest = max(longest, run_length)
    return longest
PY
cat >test_streak_fast.py <<'PY'
import random
from datetime import date, timedelta
 
from streak import longest_streak_ever
from streak_fast import longest_streak_ever_fast
 
 
def _random_history(n, seed):
    rng = random.Random(seed)
    start = date(2000, 1, 1)
    return {start + timedelta(days=rng.randrange(0, n * 3)) for _ in range(n)}
 
 
def test_naive_and_fast_agree_on_twenty_random_histories():
    for seed in range(20):
        dates = _random_history(500, seed)
        assert longest_streak_ever(dates) == longest_streak_ever_fast(dates)
PY
git add streak_fast.py test_streak_fast.py
git commit -q -m "perf(capstone): replace the sort-based streak scan with an O(n) ordinal scan"
"$PYTEST" -q
echo "commit 3: green -- now measuring the real, live speedup on THIS machine:"
"$CODE_DIR/.venv/bin/python" - <<'PYEOF'
import random
import time
from datetime import date, timedelta
 
import sys
 
sys.path.insert(0, ".")
from streak import longest_streak_ever
from streak_fast import longest_streak_ever_fast
 
rng = random.Random(7)
start = date(2000, 1, 1)
dates = {start + timedelta(days=rng.randrange(0, 150_000)) for _ in range(50_000)}
 
t0 = time.perf_counter()
naive_result = longest_streak_ever(dates)
t1 = time.perf_counter()
fast_result = longest_streak_ever_fast(dates)
t2 = time.perf_counter()
 
assert naive_result == fast_result, "algorithms disagree -- would NOT ship this"
naive_s = t1 - t0
fast_s = t2 - t1
print(f"n=50000: naive {naive_s:.6f}s, fast {fast_s:.6f}s, {naive_s / fast_s:.2f}x")
PYEOF
 
echo
echo "=== COMMIT 4/4: Step 4 -- a local lint+test gate, added and proven ==="
cat >gate.sh <<'SH'
#!/usr/bin/env bash
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
cd "$SCRIPT_DIR"
echo "lint:"
"$1" check .
echo "test:"
"$2" -q
SH
chmod +x gate.sh
git add gate.sh
git commit -q -m "docs(capstone): add a local lint+test gate mirroring the pipeline's own two stages"
"$RUFF" check . --quiet
bash gate.sh "$RUFF" "$PYTEST"
echo "commit 4: green"
 
echo
echo "=== clean history (4 commits) ==="
git log --oneline "$BASE"..HEAD
 
echo
echo "=== deriving the SemVer bump and changelog entry from v1.0.0..HEAD (topic 30 co-03/co-04) ==="
COMMITS=$(git log --pretty=%s v1.0.0..HEAD)
echo "$COMMITS"
 
HIGHEST="PATCH"
if echo "$COMMITS" | grep -Eq '^[a-z]+(\(.+\))?!:|BREAKING CHANGE:'; then
	HIGHEST="MAJOR"
elif echo "$COMMITS" | grep -Eq '^feat(\(.+\))?:'; then
	HIGHEST="MINOR"
fi
echo "highest-severity commit type present -> SemVer bump: $HIGHEST (v1.0.0 -> v1.1.0)"

$ bash scripts/build_commit_history_demo.sh
=== COMMIT 1/4: Step 1 -- import the Pass-1 baseline under a green test ===
..                                                                       [100%]
2 passed in 0.07s
commit 1: green
 
=== COMMIT 2/4: Step 2 -- SOLID/DIP: a repository Protocol + a fake, zero inheritance ===
...                                                                      [100%]
3 passed in 0.07s
commit 2: green
 
=== COMMIT 3/4: Step 3 -- O(n) algorithm, measured against the Step 1 baseline ===
....                                                                     [100%]
4 passed in 0.07s
commit 3: green -- now measuring the real, live speedup on THIS machine:
n=50000: naive 0.016162s, fast 0.005729s, 2.82x
 
=== COMMIT 4/4: Step 4 -- a local lint+test gate, added and proven ===
lint:
All checks passed!
test:
....                                                                     [100%]
4 passed in 0.08s
commit 4: green
 
=== clean history (4 commits) ===
cf411aa docs(capstone): add a local lint+test gate mirroring the pipeline's own two stages
0930717 perf(capstone): replace the sort-based streak scan with an O(n) ordinal scan
3e1130d refactor(capstone): introduce a HabitRepository DIP port and prove OCP with a fake
29adf94 feat(capstone): import the pass-1 longest-streak algorithm under a green test
 
=== deriving the SemVer bump and changelog entry from v1.0.0..HEAD (topic 30 co-03/co-04) ===
docs(capstone): add a local lint+test gate mirroring the pipeline's own two stages
perf(capstone): replace the sort-based streak scan with an O(n) ordinal scan
refactor(capstone): introduce a HabitRepository DIP port and prove OCP with a fake
feat(capstone): import the pass-1 longest-streak algorithm under a green test
highest-severity commit type present -> SemVer bump: MINOR (v1.0.0 -> v1.1.0)