diff --git a/CLAUDE.md b/CLAUDE.md index 61cc268..201b63b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -4,79 +4,69 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co ## Commands -```bash -just install # lock + sync all deps (run after pulling or changing pyproject.toml) -just lint # eof-fixer, ruff format, ruff check --fix, ty check -just lint-ci # same but no auto-fix (used in CI) -just build # build the application Docker image -just test # run all tests in Docker (starts Redpanda, runs pytest, tears down) -just test-branch # same with branch coverage -just down # tear down all containers -just publish # bump version to $GITHUB_REF_NAME, build, publish to PyPI -``` +`just --list` is the source of truth; run `just install` after pulling or +changing `pyproject.toml`. Non-obvious notes: -Run a single test file or test by name: -```bash -uv run --no-sync pytest tests/test_kafka_committer.py -uv run --no-sync pytest -k test_committer_logs_task_exceptions -``` +- `just test` / `just test-branch` run in Docker (start Redpanda, run pytest, + tear down) — they require Docker. +- Run one test without Docker via the already-running stack: + `uv run --no-sync pytest tests/test_kafka_committer.py -k `. +- `just lint` auto-fixes; `just lint-ci` is the check-only CI variant (and runs + the planning validator). ## Workflow -Per-feature: brainstorming → spec in `planning/changes/YYYY-MM-DD.NN-/design.md` → writing-plans → plan in `planning/changes/YYYY-MM-DD.NN-/plan.md` → executing-plans / subagent-driven-development → requesting-code-review → finishing-a-development-branch. Each change is a folder bundle; `` is a kebab-case description, not a story ID; `.NN` is a zero-padded intra-day counter that breaks same-date ties so the timeline sorts stably. The implementing PR sets `status: shipped` and fills `pr` / `outcome` in the branch, alongside the code and promotes its conclusions into the affected `architecture/.md` — that hand-edit keeps `architecture/` true and is the only ship-time step; there is no folder move. The change listing is generated — run `just index`. A design decision taken without a code change — especially a candidate rejected with a load-bearing reason — is recorded as `planning/decisions/YYYY-MM-DD-.md` (the `decision.md` template, frontmatter `status: accepted|superseded`), each with a **Revisit trigger** so future reviews don't re-litigate it; listed by `just index`. See [`planning/README.md`](planning/README.md) for the conventions and [`planning/_templates/`](planning/_templates/) for copy-and-fill starting points. - -**Spec** (`design.md`) captures the *thinking* — why, what the design is, trade-offs, scope. Written before code; rarely revised after merge. **Plan** (`plan.md`) captures the *sequencing* — the ordered checklist an executor walks; references the spec for the "why". **`architecture/`** captures the *invariants* of shipped systems — the living truth, promoted in the implementing PR alongside the code. A plan paragraph that would still read correctly with all task numbers and checkboxes removed is design content and belongs in the spec. - -**Three lanes.** Scale the artifact to the change. **Full** — a `design.md` + `plan.md` bundle — for real design judgment, a new file/module, a public-API change, cross-cutting/multi-file work, or non-trivial test design. **Lightweight** — a single `change.md` — for small-but-real changes (≲30 LOC net, ≤2 files, no new file, no public-API change, a single straightforward test). **Tiny** — no bundle, just a conventional commit — for a typo, dep bump, linter/formatter/CI tweak, a mechanical rename, or a single-line config change. Heavier lane wins on ambiguity; a `change.md` that outgrows its lane splits into `design.md` + `plan.md`. - -Release notes are written when cutting a release (not per-feature): copy `planning/releases/TEMPLATE.md` to `planning/releases/.md` (bare version, no `v` prefix) and link back to the driving change bundle under `planning/changes/`. +Planning uses a portable two-axis convention: `architecture/` (repo root) is the +living **truth home** and promotion target; `planning/changes/` holds the +per-change bundles. **Start at the Quick path** in +[`planning/README.md`](planning/README.md) to choose a lane (Full / Lightweight +/ Tiny), create a bundle, and ship — that file is the authoritative spec. Run +`just check-planning` to validate bundles and `just index` to print the listing. +Release notes: copy `planning/releases/TEMPLATE.md` to +`planning/releases/.md` (bare version, no `v` prefix) when cutting a +release. ## Architecture -The library provides concurrent Kafka message processing for FastStream. Modules: - -**`processing.py` — `KafkaConcurrentHandler`** -The core engine. One handler is created per `initialize_concurrent_processing` call and stored in FastStream's `ContextRepo` under the key `"concurrent_processing"`. It is *not* a singleton — calling `stop_concurrent_processing` clears the context entry so a fresh handler can be initialised. The handler manages: -- An `asyncio.Semaphore` for concurrency limiting (minimum: 1) -- A `set[asyncio.Task]` (`_tracked_tasks`) holding in-flight user tasks; the per-task done-callback (`_finish_task`) releases the semaphore and removes the task from the set -- A `KafkaBatchCommitter` for offset commits - -Key design: `handle_task()` fires-and-forgets the user coroutine as an asyncio task and enqueues a `KafkaCommitTask` on the committer. Offsets are not committed until the user task finishes (at-least-once semantics). - -`stop()` cancels every in-flight tracked task, then awaits `committer.close()`. The committer treats cancelled tasks as a hard offset boundary (see `batch_committer.py`), so cancelled-and-after offsets stay uncommitted and get redelivered on restart. Total wall-clock for shutdown is bounded by the committer's own `shutdown_timeout_sec` (default 20 s) and is sub-second in normal conditions. The handler does *not* install signal handlers — shutdown is driven by the FastStream lifespan calling `stop_concurrent_processing`. - -**`middleware.py` — FastStream middleware + lifecycle functions** -- `KafkaConcurrentProcessingMiddleware`: FastStream `BaseMiddleware` subclass. Its `consume_scope` retrieves the handler from `self.context`. It passes through (a) FakeConsumer (TestKafkaBroker) and (b) any subscriber whose ack policy is not MANUAL (`kafka_message.committed is not None`). It refuses if `_enable_auto_commit=True` on the consumer. If the handler has been stopped, it logs a warning and skips the message (the offset stays uncommitted, so the message is redelivered on restart). -- `initialize_concurrent_processing(context, ...)`: create and start a handler, store it in context. -- `stop_concurrent_processing(context)`: gates on `is_running`; calls `handler.stop()` and clears the context entry. Safe to call when the committer task has already died — `KafkaBatchCommitter.close()` early-returns on a `done()` task and logs any exception. - -**`healthcheck.py` — `is_kafka_handler_healthy`** -A single function that accepts a `ContextRepo` and returns `True` if the handler is present and `is_healthy` (i.e. `_is_running` AND committer task alive). Intended for readiness/liveness probes. - -**`batch_committer.py` — `KafkaBatchCommitter`** -Runs as a background asyncio task (`spawn()`). Streaming loop: continuously absorbs `KafkaCommitTask`s from the queue into per-partition pending state, watches the head not-done task per partition, and commits each partition's contiguous-done prefix when total pending ≥ `commit_batch_size`, when `commit_batch_timeout_sec` fires, or when `commit_all`/`close` sets the flush event. Per partition, `_extract_ready_prefixes` sorts by offset (tolerates re-queued tasks landing out of order) and stops at the first not-done task; a cancelled task is a hard boundary — cancelled + everything after is dropped from pending while `_map_offsets_per_partition` stops the offset advance at the cancelled task (so uncommitted offsets get redelivered on restart, at-least-once). Per consumer-id group, commits via `consumer.commit({TopicPartition: max_offset+1})`. Transient `KafkaError` re-queues the batch; `CommitFailedError`/`IllegalStateError` (rebalance/revocation) discards it. `CommitterIsDeadError` is raised to callers when the committer's main task has died, which triggers `handler.stop()`. - -**`rebalance.py` — `ConsumerRebalanceListener`** -Returned by `handler.create_rebalance_listener()`. On `on_partitions_revoked`, calls `committer.commit_all()` so offsets are flushed before the partition is reassigned, preventing duplicate processing after rebalance. - -## Key patterns +The library provides concurrent Kafka message processing for FastStream. The +authoritative, code-current account of each capability lives in +[`architecture/`](architecture/). **When a change alters a capability's +behavior, update the matching `architecture/.md` in the same PR** — +that promotion is what keeps `architecture/` true. + +Invariants (what must not break): + +- **At-least-once offsets.** Offsets are committed only *after* the user task + finishes; a crash, cancellation, or rebalance before completion leaves the + offset uncommitted so the message is redelivered. A cancelled task is a hard + offset boundary — cancelled-and-after offsets on its partition stay + uncommitted. +- **One handler per init, not a singleton.** A handler lives in `ContextRepo` + under `"concurrent_processing"`; `stop_concurrent_processing` clears it so a + fresh handler can be initialised. Lifecycle is owned by whoever calls + init/stop — no module-level state, no signal handlers. +- **Middleware gates on manual acks.** It passes through FakeConsumer and + non-MANUAL-ack subscribers, refuses `_enable_auto_commit=True`, rejects batch + subscribers, and skips (logs, leaves offset uncommitted) once the handler is + stopped. The `_classify(...) -> _Route` branch *order* is load-bearing. +- **Bounded shutdown / rebalance flush.** Shutdown is bounded by the committer's + `shutdown_timeout_sec` (default 20 s); the rebalance listener's `commit_all` + flush is bounded by `flush_timeout_sec` (default 10 s, well under aiokafka's + 300 s `max.poll.interval.ms`). +- **Real-broker tests.** Integration tests drive a real Redpanda container; the + FastStream/aiokafka harness invariants (start subscribers explicitly, + `auto_offset_reset="earliest"`, pre-create topics, etc.) are load-bearing. + +| Capability | File | +|---|---| +| `KafkaConcurrentHandler` (`processing.py`) — engine, dispatch, shutdown | [`architecture/concurrent-handler.md`](architecture/concurrent-handler.md) | +| `KafkaBatchCommitter` (`batch_committer.py`) — offset-commit task | [`architecture/batch-committer.md`](architecture/batch-committer.md) | +| Middleware, init/stop lifecycle, healthcheck (`middleware.py`, `healthcheck.py`) | [`architecture/middleware-lifecycle.md`](architecture/middleware-lifecycle.md) | +| `ConsumerRebalanceListener` (`rebalance.py`) | [`architecture/rebalance.md`](architecture/rebalance.md) | +| Real-broker integration-test harness (`tests/test_integration.py`) | [`architecture/integration-tests.md`](architecture/integration-tests.md) | + +## Conventions - **Type suppression**: use `# ty: ignore[rule-name]` (not `# type: ignore`). - **No `from __future__ import annotations`**: annotations are evaluated eagerly; `typing.Self`/`typing.Never` are used directly (requires Python ≥ 3.11). - **Imports at module level**: no local imports inside functions. - -## Integration tests - -`tests/test_integration.py` runs against a real Redpanda container (Kafka-compatible, lightweight) via `testcontainers[kafka]`. The container is session-scoped — one instance for the whole test run. - -**Running integration tests** requires Docker — they run automatically as part of `just test`. - -**Key findings from building these tests:** - -- `async with KafkaBroker():` only calls `connect()`, which sets up the producer. It does **not** start subscribers. You must also call `await broker.start()` explicitly to launch the consumer poll tasks. -- Always use `auto_offset_reset="earliest"` on test subscribers. The default `"latest"` causes the consumer to miss messages published before it gets its partition assignment. -- Pre-create topics with `AIOKafkaAdminClient` before starting the broker. Auto-creation on first publish triggers a `NotLeaderForPartitionError` retry loop that can outlast short sleeps. -- After `await broker.start()`, sleep ~1.5 s before publishing to let the consumer join the group and receive partition assignments. -- `AsgiFastStream` lifespan tests must use `async with app.start_lifespan_context()` — calling `app.start()` / `app.stop()` bypasses the `lifespan` context manager entirely. -- `AsgiFastStream` injects its own app-level `ContextRepo` into the lifespan, separate from `broker.context`. Pass `broker.context` explicitly to `initialize_concurrent_processing` and `stop_concurrent_processing`. -- Subscriber-level `middlewares` on `@broker.subscriber(...)` takes `SubscriberMiddleware` (a plain `(call_next, msg)` callable), not `BaseMiddleware` subclasses. To scope `KafkaConcurrentProcessingMiddleware` to a subset of subscribers, use `KafkaRouter(middlewares=[KafkaConcurrentProcessingMiddleware])` and `broker.include_router(router)`. diff --git a/Justfile b/Justfile index 2382e61..8ff25de 100644 --- a/Justfile +++ b/Justfile @@ -27,11 +27,16 @@ lint-ci: uv run ruff format --check uv run ruff check --no-fix uv run ty check + uv run python planning/index.py --check -# Print the planning change index (grouped by status) to stdout. +# Print the planning change index (flat, newest-first) to stdout. index: uv run python planning/index.py +# Validate planning bundles + decisions; CI runs this. +check-planning: + uv run python planning/index.py --check + publish: rm -rf dist uv version $GITHUB_REF_NAME diff --git a/architecture/README.md b/architecture/README.md new file mode 100644 index 0000000..cf1ab20 --- /dev/null +++ b/architecture/README.md @@ -0,0 +1,35 @@ +# Architecture + +The living truth about what `faststream-concurrent-aiokafka` does **now** — one +file per capability, updated by hand whenever a change ships. The *why* and *how +it got here* live in [`../planning/changes/`](../planning/changes/), and +decisions deliberately taken (including options rejected) in +[`../planning/decisions/`](../planning/decisions/); this directory is the present. + +Each capability file is an **implementation-detail** page. Its terse +**invariant summary** ("what must not break") lives in +[`../CLAUDE.md`](../CLAUDE.md) § Architecture. + +These files carry **no frontmatter** — they are prose, dated by git. + +## Capabilities + +- [concurrent-handler.md](concurrent-handler.md) — `KafkaConcurrentHandler`, the + core engine: semaphore-bounded concurrency, task tracking, fire-and-forget + dispatch, and bounded shutdown. +- [batch-committer.md](batch-committer.md) — `KafkaBatchCommitter`, the + background offset-commit task: per-partition contiguous-done prefixes, + cancelled-task boundaries, and at-least-once redelivery. +- [middleware-lifecycle.md](middleware-lifecycle.md) — the FastStream + middleware, `initialize_concurrent_processing` / + `stop_concurrent_processing`, and the healthcheck. +- [rebalance.md](rebalance.md) — the `ConsumerRebalanceListener` that flushes + offsets on partition revocation. +- [integration-tests.md](integration-tests.md) — the real-broker (Redpanda) + test harness and its load-bearing FastStream/aiokafka driving invariants. + +## Promotion rule + +Shipping a change hand-edits the affected capability file(s) here to match the +new reality, in the same PR as the code. The change bundle stays in place under +[`../planning/changes/`](../planning/changes/) — no folder move. diff --git a/architecture/integration-tests.md b/architecture/integration-tests.md new file mode 100644 index 0000000..566bd0d --- /dev/null +++ b/architecture/integration-tests.md @@ -0,0 +1,40 @@ +# Integration tests + +`tests/test_integration.py` exercises the concurrent handler against a **real** +Kafka-compatible broker rather than mocks, so the offset-commit and rebalance +behaviour is verified end to end. + +## Broker + +Tests run against a Redpanda container (Kafka-compatible, lightweight) via +`testcontainers[kafka]`. The container is **session-scoped** — one instance for +the whole test run — and therefore requires Docker. It runs automatically as +part of `just test`. + +## Harness invariants + +These are load-bearing facts about driving FastStream/aiokafka in tests; getting +any of them wrong produces flaky or silently-empty test runs: + +- `async with KafkaBroker():` only calls `connect()`, which sets up the + producer. It does **not** start subscribers — you must also call + `await broker.start()` explicitly to launch the consumer poll tasks. +- Always use `auto_offset_reset="earliest"` on test subscribers. The default + `"latest"` makes the consumer miss messages published before it gets its + partition assignment. +- Pre-create topics with `AIOKafkaAdminClient` before starting the broker. + Auto-creation on first publish triggers a `NotLeaderForPartitionError` retry + loop that can outlast short sleeps. +- After `await broker.start()`, sleep ~1.5 s before publishing to let the + consumer join the group and receive partition assignments. +- `AsgiFastStream` lifespan tests must use + `async with app.start_lifespan_context()` — calling `app.start()` / + `app.stop()` bypasses the `lifespan` context manager entirely. +- `AsgiFastStream` injects its own app-level `ContextRepo` into the lifespan, + separate from `broker.context`. Pass `broker.context` explicitly to + `initialize_concurrent_processing` and `stop_concurrent_processing`. +- Subscriber-level `middlewares` on `@broker.subscriber(...)` takes a + `SubscriberMiddleware` (a plain `(call_next, msg)` callable), not a + `BaseMiddleware` subclass. To scope `KafkaConcurrentProcessingMiddleware` to a + subset of subscribers, use `KafkaRouter(middlewares=[...])` plus + `broker.include_router(router)`. diff --git a/planning/.convention-version b/planning/.convention-version new file mode 100644 index 0000000..3eefcb9 --- /dev/null +++ b/planning/.convention-version @@ -0,0 +1 @@ +1.0.0 diff --git a/planning/README.md b/planning/README.md index d05711f..892d2cb 100644 --- a/planning/README.md +++ b/planning/README.md @@ -33,10 +33,11 @@ A change is a folder `changes/YYYY-MM-DD.NN-/`: (`.01`, `.02`, …) that breaks same-date ties so the timeline sorts stably. - `` — kebab-case description, not a story ID. -`summary` is written when the change is created (it is the change's -one-liner). The implementing PR then sets `status: shipped` and fills `pr` -and `outcome` **in the branch**, alongside the code and the `architecture/` -promotion — no post-merge bookkeeping, no folder move. +`summary` is written when the change is created (the intent one-liner) and +**finalized at ship** to state the realized result — set in the implementing +PR, alongside the code and the `architecture/` promotion. No post-merge +bookkeeping, no folder move. `date` and `slug` are never written — they are +read from the bundle's directory name. ### Three lanes @@ -67,19 +68,25 @@ Templates live in [`_templates/`](_templates/). ### Frontmatter -`design.md` / `change.md`: `status` (draft|approved|shipped|superseded), -`date`, `slug`, `summary` (single line), `supersedes`, `superseded_by`, `pr`, -`outcome`. `plan.md`: `status`, `date`, `slug`, `spec`, `pr`. -`decisions/*.md`: `status` (accepted|superseded), `date`, `slug`, `summary`, -`supersedes`, `superseded_by`, `pr`. Files in -`architecture/` carry **no** frontmatter — living prose, dated by git. +`date` and `slug` are **derived from the directory / file name** — never +repeated in frontmatter. So: + +- `design.md` / `change.md`: `summary` (single line) only. +- `plan.md`: **no frontmatter** — its identity is the bundle directory. +- `decisions/*.md`: `status` (accepted|superseded), `summary`, and optional + `supersedes` / `superseded_by`. +- Files in `architecture/` carry **no** frontmatter — living prose, dated by git. + +**`summary`** is one line: written at creation as the intent, then **finalized +at ship** to state the realized result — what shipped and its effect. It is the +only field the index renders. ## Index The listing is **generated**, not maintained — run `just index` to print it: -changes grouped by `status` (In progress / Shipped / Superseded), then -decisions newest-first. The frontmatter in each bundle / decision file is the -single source of truth; there is no committed copy to drift. +changes newest-first, then decisions newest-first. The `summary` in each +bundle / decision file is the single source of truth; there is no committed +copy to drift. ## Other diff --git a/planning/_templates/change.md b/planning/_templates/change.md index 7ffec26..d4c8962 100644 --- a/planning/_templates/change.md +++ b/planning/_templates/change.md @@ -1,12 +1,5 @@ --- -status: draft -date: YYYY-MM-DD -slug: my-change -summary: One line — shown in the generated index. Fill at ship time. -supersedes: null -superseded_by: null -pr: null -outcome: null +summary: One line — shown in the generated index. Written at creation; finalize at ship to state the realized result. --- # Change: One-line capitalized title diff --git a/planning/_templates/decision.md b/planning/_templates/decision.md index 940fb37..45ccaf0 100644 --- a/planning/_templates/decision.md +++ b/planning/_templates/decision.md @@ -1,11 +1,8 @@ --- status: accepted # accepted | superseded -date: YYYY-MM-DD -slug: my-decision summary: One line — shown in `just index`. supersedes: null superseded_by: null -pr: null # PR/commit where the decision was made or recorded --- # One-line capitalized title diff --git a/planning/_templates/design.md b/planning/_templates/design.md index b9e11c9..d63e22d 100644 --- a/planning/_templates/design.md +++ b/planning/_templates/design.md @@ -1,12 +1,5 @@ --- -status: draft -date: YYYY-MM-DD -slug: my-change -summary: One line — shown in the generated index. Fill at ship time. -supersedes: null -superseded_by: null -pr: null -outcome: null +summary: One line — shown in the generated index. Written at creation; finalize at ship to state the realized result. --- # Design: One-line capitalized title diff --git a/planning/_templates/plan.md b/planning/_templates/plan.md index f2b90e8..132d720 100644 --- a/planning/_templates/plan.md +++ b/planning/_templates/plan.md @@ -1,11 +1,3 @@ ---- -status: draft -date: YYYY-MM-DD -slug: my-change -spec: my-change -pr: null ---- - # — implementation plan > **For agentic workers:** REQUIRED SUB-SKILL: Use @@ -46,9 +38,7 @@ in the spec. ```bash git add path/to/file.py - git commit -m ": - - Co-Authored-By: Claude Opus 4.7 (1M context) " + git commit -m ": " ``` --- diff --git a/planning/_templates/release.md b/planning/_templates/release.md new file mode 100644 index 0000000..5081187 --- /dev/null +++ b/planning/_templates/release.md @@ -0,0 +1,38 @@ +# + + + + + +## Feature + +- **.** What it adds and how to use it. + +## Fix + +- **.** What was broken, now fixed (reference the issue/regression). + +## Internal refactors + +- **.** What changed under the hood, stated as no behavior change. + +## Packaging + +- Metadata / build / dependency changes visible to installers. + +## Why + +Context a reader needs for the headline change. Omit for small releases. + +## Downstream + +What dependents must do — e.g. bump their version floor — or "No action +needed" when there is no API change. Omit if the project has no downstreams. + +## Internals + +- Coverage / tooling notes. diff --git a/planning/changes/2026-06-03.01-faststream-0.7-migration/design.md b/planning/changes/2026-06-03.01-faststream-0.7-migration/design.md index a0b9fb1..9553ec6 100644 --- a/planning/changes/2026-06-03.01-faststream-0.7-migration/design.md +++ b/planning/changes/2026-06-03.01-faststream-0.7-migration/design.md @@ -1,12 +1,5 @@ --- -status: shipped -date: 2026-06-03 -slug: faststream-0.7-migration summary: Migrate to `faststream>=0.7` (drop 0.6 support). Design-only bundle (execution plan removed post-merge). -supersedes: null -superseded_by: null -pr: "28" -outcome: "merged as #28 (plan removed post-execution; design-only bundle)" --- # Design: FastStream 0.7 migration (two-PR split) diff --git a/planning/changes/2026-06-04.01-faststream-0.7.1-testbroker-typing/design.md b/planning/changes/2026-06-04.01-faststream-0.7.1-testbroker-typing/design.md index d5c33c7..ef70cdd 100644 --- a/planning/changes/2026-06-04.01-faststream-0.7.1-testbroker-typing/design.md +++ b/planning/changes/2026-06-04.01-faststream-0.7.1-testbroker-typing/design.md @@ -1,12 +1,5 @@ --- -status: shipped -date: 2026-06-04 -slug: faststream-0.7.1-testbroker-typing summary: Adopt FastStream 0.7.1's `TestBroker` typing fix; drop `# ty: ignore` directives. -supersedes: null -superseded_by: null -pr: "29" -outcome: "merged as #29" --- # FastStream 0.7.1 TestBroker typing alignment — design diff --git a/planning/changes/2026-06-04.01-faststream-0.7.1-testbroker-typing/plan.md b/planning/changes/2026-06-04.01-faststream-0.7.1-testbroker-typing/plan.md index 04497d1..25e9827 100644 --- a/planning/changes/2026-06-04.01-faststream-0.7.1-testbroker-typing/plan.md +++ b/planning/changes/2026-06-04.01-faststream-0.7.1-testbroker-typing/plan.md @@ -1,11 +1,3 @@ ---- -status: shipped -date: 2026-06-04 -slug: faststream-0.7.1-testbroker-typing -spec: faststream-0.7.1-testbroker-typing -pr: "29" ---- - # FastStream 0.7.1 TestBroker typing alignment — implementation plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. diff --git a/planning/changes/2026-06-13.01-robustness-docs-test-audit/design.md b/planning/changes/2026-06-13.01-robustness-docs-test-audit/design.md index e21ee24..28be4af 100644 --- a/planning/changes/2026-06-13.01-robustness-docs-test-audit/design.md +++ b/planning/changes/2026-06-13.01-robustness-docs-test-audit/design.md @@ -1,12 +1,5 @@ --- -status: shipped -date: 2026-06-13 -slug: robustness-docs-test-audit summary: Rebalance-flush timeout, batch-subscriber guard, uncommitted-task backpressure, plus docs/test/refactor. Shipped in 0.6.0. -supersedes: null -superseded_by: null -pr: "32" -outcome: "merged as #32; shipped in 0.6.0" --- # Robustness, docs, and test audit — design diff --git a/planning/changes/2026-06-13.01-robustness-docs-test-audit/plan.md b/planning/changes/2026-06-13.01-robustness-docs-test-audit/plan.md index 6a75fc4..4e149a2 100644 --- a/planning/changes/2026-06-13.01-robustness-docs-test-audit/plan.md +++ b/planning/changes/2026-06-13.01-robustness-docs-test-audit/plan.md @@ -1,11 +1,3 @@ ---- -status: shipped -date: 2026-06-13 -slug: robustness-docs-test-audit -spec: robustness-docs-test-audit -pr: "32" ---- - # Robustness, Docs, and Test Audit — Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. diff --git a/planning/changes/2026-06-13.02-codify-release-notes/design.md b/planning/changes/2026-06-13.02-codify-release-notes/design.md index 7a0562d..e681d45 100644 --- a/planning/changes/2026-06-13.02-codify-release-notes/design.md +++ b/planning/changes/2026-06-13.02-codify-release-notes/design.md @@ -1,12 +1,5 @@ --- -status: shipped -date: 2026-06-13 -slug: codify-release-notes summary: Codify `planning/releases/` as a workflow step; add the release-notes template. -supersedes: null -superseded_by: null -pr: "33" -outcome: "merged as #33" --- # Codify `planning/releases/` as a workflow step diff --git a/planning/changes/2026-06-13.02-codify-release-notes/plan.md b/planning/changes/2026-06-13.02-codify-release-notes/plan.md index ac14262..5d4f2a4 100644 --- a/planning/changes/2026-06-13.02-codify-release-notes/plan.md +++ b/planning/changes/2026-06-13.02-codify-release-notes/plan.md @@ -1,11 +1,3 @@ ---- -status: shipped -date: 2026-06-13 -slug: codify-release-notes -spec: codify-release-notes -pr: "33" ---- - # Codify `planning/releases/` as a Workflow Step — Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. diff --git a/planning/changes/2026-06-13.03-portable-planning-convention/design.md b/planning/changes/2026-06-13.03-portable-planning-convention/design.md index 019370a..d4ad5f3 100644 --- a/planning/changes/2026-06-13.03-portable-planning-convention/design.md +++ b/planning/changes/2026-06-13.03-portable-planning-convention/design.md @@ -1,12 +1,5 @@ --- -status: shipped -date: 2026-06-13 -slug: portable-planning-convention summary: Adopt the portable two-axis convention: `architecture/` truth home + `changes/` bundles, fresh Index. -supersedes: null -superseded_by: null -pr: "34" -outcome: "ships in #34; defines the convention, no architecture/ promotion applies" --- # Design: Adopt the portable OpenSpec-shaped planning convention diff --git a/planning/changes/2026-06-13.03-portable-planning-convention/plan.md b/planning/changes/2026-06-13.03-portable-planning-convention/plan.md index 5933eb9..4a5545b 100644 --- a/planning/changes/2026-06-13.03-portable-planning-convention/plan.md +++ b/planning/changes/2026-06-13.03-portable-planning-convention/plan.md @@ -1,11 +1,3 @@ ---- -status: shipped -date: 2026-06-13 -slug: portable-planning-convention -spec: portable-planning-convention -pr: "34" ---- - # Portable planning convention — implementation plan > **For agentic workers:** REQUIRED SUB-SKILL: Use diff --git a/planning/changes/2026-06-23.01-pending-commits-deep-module/design.md b/planning/changes/2026-06-23.01-pending-commits-deep-module/design.md index 5a4d6a3..c585e2b 100644 --- a/planning/changes/2026-06-23.01-pending-commits-deep-module/design.md +++ b/planning/changes/2026-06-23.01-pending-commits-deep-module/design.md @@ -1,12 +1,5 @@ --- -status: shipped -date: 2026-06-23 -slug: pending-commits-deep-module summary: Extract pending/watermark state into a synchronous PendingCommits module; delete the committer's static test-handle delegators. -supersedes: null -superseded_by: null -pr: "39" -outcome: "Merged as #39 (squash 3167489). Full suite green on real Redpanda in CI across Python 3.11–3.14 at 100% coverage." --- # Design: Give pending-commit bookkeeping a home (`PendingCommits`) diff --git a/planning/changes/2026-06-23.01-pending-commits-deep-module/plan.md b/planning/changes/2026-06-23.01-pending-commits-deep-module/plan.md index 6b118ee..e9cf381 100644 --- a/planning/changes/2026-06-23.01-pending-commits-deep-module/plan.md +++ b/planning/changes/2026-06-23.01-pending-commits-deep-module/plan.md @@ -1,11 +1,3 @@ ---- -status: shipped -date: 2026-06-23 -slug: pending-commits-deep-module -spec: pending-commits-deep-module -pr: "39" ---- - # pending-commits-deep-module — implementation plan > **For agentic workers:** REQUIRED SUB-SKILL: Use diff --git a/planning/changes/2026-06-24.01-commit-scheduler-deep-module/design.md b/planning/changes/2026-06-24.01-commit-scheduler-deep-module/design.md index 7d7e44d..76b88fa 100644 --- a/planning/changes/2026-06-24.01-commit-scheduler-deep-module/design.md +++ b/planning/changes/2026-06-24.01-commit-scheduler-deep-module/design.md @@ -1,12 +1,5 @@ --- -status: shipped -date: 2026-06-24 -slug: commit-scheduler-deep-module summary: Extract the streaming loop's when-to-commit decision state into a pure synchronous CommitScheduler; the committer keeps the async driver. -supersedes: null -superseded_by: null -pr: "40" -outcome: "Merged as #40 (squash d3aa133). Full suite green on real Redpanda in CI across Python 3.11–3.14 at 100% coverage." --- # Design: Extract the commit-timing decisions into `CommitScheduler` diff --git a/planning/changes/2026-06-24.01-commit-scheduler-deep-module/plan.md b/planning/changes/2026-06-24.01-commit-scheduler-deep-module/plan.md index 3ec1090..29540f5 100644 --- a/planning/changes/2026-06-24.01-commit-scheduler-deep-module/plan.md +++ b/planning/changes/2026-06-24.01-commit-scheduler-deep-module/plan.md @@ -1,11 +1,3 @@ ---- -status: shipped -date: 2026-06-24 -slug: commit-scheduler-deep-module -spec: commit-scheduler-deep-module -pr: "40" ---- - # commit-scheduler-deep-module — implementation plan > **For agentic workers:** REQUIRED SUB-SKILL: Use diff --git a/planning/changes/2026-06-24.02-route-classify-extraction/change.md b/planning/changes/2026-06-24.02-route-classify-extraction/change.md index 77adb11..0dc5d01 100644 --- a/planning/changes/2026-06-24.02-route-classify-extraction/change.md +++ b/planning/changes/2026-06-24.02-route-classify-extraction/change.md @@ -1,12 +1,5 @@ --- -status: shipped -date: 2026-06-24 -slug: route-classify-extraction summary: Extract consume_scope's 8-branch message classification into a pure _classify(...) -> _Route; consume_scope becomes a precondition + match. -supersedes: null -superseded_by: null -pr: "41" -outcome: "Merged as #41 (squash 596b2d7). Full suite green on real Redpanda in CI across Python 3.11–3.14; middleware.py at 100% coverage. PR review added an assert_never exhaustiveness guard + test hardenings, no correctness findings." --- # Change: Lift the route decision out of `consume_scope` diff --git a/planning/decisions/2026-06-24-batch-subscribers-unsupported.md b/planning/decisions/2026-06-24-batch-subscribers-unsupported.md index 4e70e43..7ef05e5 100644 --- a/planning/decisions/2026-06-24-batch-subscribers-unsupported.md +++ b/planning/decisions/2026-06-24-batch-subscribers-unsupported.md @@ -1,11 +1,6 @@ --- status: accepted -date: 2026-06-24 -slug: batch-subscribers-unsupported summary: Batch subscribers (batch=True) are deliberately unsupported; the middleware rejects them with a clear RuntimeError. -supersedes: null -superseded_by: null -pr: null --- # Batch subscribers (`batch=True`) are unsupported diff --git a/planning/decisions/2026-06-24-health-delegation-chain-not-deepened.md b/planning/decisions/2026-06-24-health-delegation-chain-not-deepened.md index 7516c2e..8310b27 100644 --- a/planning/decisions/2026-06-24-health-delegation-chain-not-deepened.md +++ b/planning/decisions/2026-06-24-health-delegation-chain-not-deepened.md @@ -1,11 +1,6 @@ --- status: accepted -date: 2026-06-24 -slug: health-delegation-chain-not-deepened summary: Leave the is_kafka_handler_healthy → handler.is_healthy → committer.is_healthy chain as-is; not worth deepening yet. -supersedes: null -superseded_by: null -pr: null --- # Health delegation chain left as-is (architecture candidate #4) diff --git a/planning/index.py b/planning/index.py index 392860a..8916661 100644 --- a/planning/index.py +++ b/planning/index.py @@ -2,24 +2,29 @@ """ Generate the planning index from frontmatter. -Run via ``just index``. Globs ``planning/changes/*/`` (each bundle's ``design.md``, -falling back to ``change.md``) and ``planning/decisions/*.md``, reads their -frontmatter, and prints a Markdown listing to stdout — changes grouped by lifecycle -status, then decisions newest-first. Never writes a file: the listing is a query over -the files, not a committed artifact. +Run via ``just index``. Globs ``planning/changes/*/`` (each bundle's +``design.md``, falling back to ``change.md``) and ``planning/decisions/*.md``, +reads their frontmatter, and prints a Markdown listing to stdout — changes +then decisions, newest-first. Never writes a file: +the listing is a query over the files, not a committed artifact. + +``date`` and ``slug`` are derived from the directory / file name, not +frontmatter — the name is the single source of truth for both. """ import pathlib +import re import sys CHANGES_DIR = pathlib.Path(__file__).parent / "changes" DECISIONS_DIR = pathlib.Path(__file__).parent / "decisions" -GROUPS: tuple[tuple[str, tuple[str, ...]], ...] = ( - ("In progress", ("draft", "approved")), - ("Shipped", ("shipped",)), - ("Superseded", ("superseded",)), -) +VALID_DECISION_STATUS = {"accepted", "superseded"} +BUNDLE_RE = re.compile(r"^(?P\d{4}-\d{2}-\d{2})\.\d{2}-(?P.+)$") +DECISION_RE = re.compile(r"^(?P\d{4}-\d{2}-\d{2})-(?P.+)$") +ALLOWED_BUNDLE_FILES = {"design.md", "plan.md", "change.md"} +SPEC_REQUIRED = ("summary",) +DECISION_REQUIRED = ("status", "summary") def parse_frontmatter(text: str) -> dict[str, str]: @@ -31,6 +36,8 @@ def parse_frontmatter(text: str) -> dict[str, str]: for line in lines[1:]: if line.strip() == "---": break + if line[:1] in (" ", "\t"): + continue key, sep, value = line.partition(": ") if not sep: continue @@ -39,8 +46,17 @@ def parse_frontmatter(text: str) -> dict[str, str]: return fields +def _named(fields: dict[str, str], name: str, pattern: re.Pattern[str]) -> dict[str, str]: + """Inject ``date``/``slug`` derived from a dir/file name into ``fields``.""" + match = pattern.match(name) + if match: + fields["date"] = match.group("date") + fields["slug"] = match.group("slug") + return fields + + def load_bundles() -> list[dict[str, str]]: - """Read every bundle's spec frontmatter under ``CHANGES_DIR``.""" + """Read each bundle's summary; derive date/slug from the directory name.""" bundles: list[dict[str, str]] = [] for bundle in sorted(CHANGES_DIR.iterdir()): if not bundle.is_dir(): @@ -50,7 +66,7 @@ def load_bundles() -> list[dict[str, str]]: spec = bundle / "change.md" if not spec.exists(): continue - fields = parse_frontmatter(spec.read_text(encoding="utf-8")) + fields = _named(parse_frontmatter(spec.read_text(encoding="utf-8")), bundle.name, BUNDLE_RE) fields["path"] = f"changes/{bundle.name}/{spec.name}" fields["name"] = bundle.name bundles.append(fields) @@ -58,14 +74,14 @@ def load_bundles() -> list[dict[str, str]]: def load_decisions() -> list[dict[str, str]]: - """Read frontmatter from every decision file under ``DECISIONS_DIR``.""" + """Read each decision's frontmatter; derive date/slug from the file name.""" decisions: list[dict[str, str]] = [] if not DECISIONS_DIR.is_dir(): return decisions for path in sorted(DECISIONS_DIR.glob("*.md")): if path.name == "README.md" or path.name.startswith("_"): continue - fields = parse_frontmatter(path.read_text(encoding="utf-8")) + fields = _named(parse_frontmatter(path.read_text(encoding="utf-8")), path.stem, DECISION_RE) fields["path"] = f"decisions/{path.name}" fields["name"] = path.stem decisions.append(fields) @@ -73,13 +89,12 @@ def load_decisions() -> list[dict[str, str]]: def format_row(bundle: dict[str, str]) -> str: - """Render one bundle or decision as a Markdown list item.""" + """Render one bundle as a Markdown list item.""" slug = bundle.get("slug", "?") path = bundle.get("path", "") - pr = bundle.get("pr") or "—" date = bundle.get("date", "") summary = bundle.get("summary") or "(no summary)" - line = f"- **[{slug}]({path})** (#{pr}, {date}) — {summary}" + line = f"- **[{slug}]({path})** ({date}) — {summary}" if bundle.get("supersedes"): line += f" _(supersedes {bundle['supersedes']})_" if bundle.get("superseded_by"): @@ -88,26 +103,85 @@ def format_row(bundle: dict[str, str]) -> str: def render(bundles: list[dict[str, str]], decisions: list[dict[str, str]]) -> str: - """Render the full Markdown listing: changes by status, then decisions.""" + """Render the full Markdown listing: changes then decisions, newest-first.""" out = ["# Planning index", "", "_Generated by `just index` — do not edit._", "", "## Changes", ""] - for title, statuses in GROUPS: - out += [f"### {title}", ""] - rows = sorted( - (b for b in bundles if b.get("status") in statuses), - key=lambda b: b.get("name", ""), - reverse=True, - ) - out += [format_row(b) for b in rows] if rows else ["_None._"] - out.append("") - out += ["## Decisions", ""] + change_rows = sorted(bundles, key=lambda b: b.get("name", ""), reverse=True) + out += [format_row(b) for b in change_rows] if change_rows else ["_None._"] + out += ["", "## Decisions", ""] decision_rows = sorted(decisions, key=lambda d: d.get("name", ""), reverse=True) out += [format_row(d) for d in decision_rows] if decision_rows else ["_None._"] out.append("") return "\n".join(out).rstrip() + "\n" +def _require(fields: dict[str, str], keys: tuple[str, ...], rel: str, violations: list[str]) -> None: + """Append a violation for each required key that is absent or empty.""" + violations.extend(f"{rel}: missing or empty frontmatter key '{key}'" for key in keys if not fields.get(key)) + + +def _check_spec_file(path: pathlib.Path, rel: str, violations: list[str]) -> None: + """Validate a design.md / change.md spec file (requires `summary`).""" + fields = parse_frontmatter(path.read_text(encoding="utf-8")) + _require(fields, SPEC_REQUIRED, rel, violations) + + +def _check_bundle(bundle: pathlib.Path, violations: list[str]) -> None: + """Validate one change bundle directory.""" + rel = f"changes/{bundle.name}" + if BUNDLE_RE.match(bundle.name) is None: + violations.append(f"{rel}: directory name is not 'YYYY-MM-DD.NN-slug'") + violations.extend( + f"{rel}/{child.name}: unexpected file in bundle (allowed: {', '.join(sorted(ALLOWED_BUNDLE_FILES))})" + for child in sorted(bundle.iterdir()) + if child.name not in ALLOWED_BUNDLE_FILES + ) + design = bundle / "design.md" + change = bundle / "change.md" + if not design.exists() and not change.exists(): + violations.append(f"{rel}: bundle has neither design.md nor change.md") + for spec_file in (design, change): + if spec_file.exists(): + _check_spec_file(spec_file, f"{rel}/{spec_file.name}", violations) + # plan.md carries no frontmatter — its identity comes from the bundle dir. + + +def _check_decision(path: pathlib.Path, violations: list[str]) -> None: + """Validate one decision file (requires `status` + `summary`).""" + rel = f"decisions/{path.name}" + if DECISION_RE.match(path.stem) is None: + violations.append(f"{rel}: file name is not 'YYYY-MM-DD-slug.md'") + fields = parse_frontmatter(path.read_text(encoding="utf-8")) + _require(fields, DECISION_REQUIRED, rel, violations) + status = fields.get("status", "") + if status and status not in VALID_DECISION_STATUS: + violations.append(f"{rel}: invalid status '{status}' (allowed: {', '.join(sorted(VALID_DECISION_STATUS))})") + + +def check() -> list[str]: + """Validate every bundle and decision; return the list of violation strings.""" + violations: list[str] = [] + for bundle in sorted(CHANGES_DIR.iterdir()): + if bundle.is_dir(): + _check_bundle(bundle, violations) + if DECISIONS_DIR.is_dir(): + for path in sorted(DECISIONS_DIR.glob("*.md")): + if path.name == "README.md" or path.name.startswith("_"): + continue + _check_decision(path, violations) + return violations + + def main() -> int: - """Print the listing to stdout.""" + """Print the listing to stdout, or validate bundles with --check.""" + if "--check" in sys.argv[1:]: + violations = check() + if violations: + sys.stderr.write(f"planning: {len(violations)} violation(s)\n") + for violation in violations: + sys.stderr.write(f" - {violation}\n") + return 1 + sys.stdout.write("planning: OK\n") + return 0 sys.stdout.write(render(load_bundles(), load_decisions())) return 0