> ## Documentation Index
> Fetch the complete documentation index at: https://gladlabs.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Topic batch service

# Topic Batch Service

**File:** `src/cofounder_agent/services/topic_batch_service.py`
**Tested by:** `src/cofounder_agent/tests/unit/services/test_topic_batch_service.py`
**Last reviewed:** 2026-07-01

## What it does

For a given niche, `TopicBatchService.run_sweep(niche_id=...)` discovers
fresh candidate topics from external sources (RSS, search, custom
plugins) AND from internal RAG sources (Claude sessions, brain
knowledge, audit events, decision log, memory files, post history),
embeds each candidate against the niche's goal vectors, pre-ranks the
top N per pool, scores the survivors with an LLM, writes a new
`topic_batches` row plus per-candidate rows (in `topic_candidates` for
external, `internal_topic_candidates` for internal), and opens the
operator approval gate. Carry-forward from the previous batch's
unpicked candidates is decay-multiplied (0.7^N at default) so old
ideas drift down naturally.

`show_batch`, `rank_batch`, `edit_winner`, `resolve_batch`,
`reject_batch` form the operator-side workflow that follows discovery.
`resolve_batch` hands the rank-1 candidate to the content pipeline by
inserting a `pipeline_tasks` row with `topic_batch_id` provenance.

Two deterministic **topic-sanity** checkpoints guard the flow
(`services/topic_sanity.py`, added 2026-07-01 after a dots-only dev.to
headline was LLM-ranked top of its batch and burned a full generation
run): `run_sweep` drops contentless candidate titles at intake (before
embedding, with one aggregated `topic_sanity_rejected` finding per
sweep), and `_handoff_to_pipeline` raises `TopicSanityError` before any
DB write if the shipping topic (operator edit wins) is contentless.

This service replaces the older `topic_proposal_service`. Shipped 2026-04-30.

## Public API

* `TopicBatchService(pool, *, site_config)` — constructor
  (`site_config` is required DI since the #272 Phase-2d migration).
* `await svc.run_sweep(niche_id) -> BatchSnapshot | None` — full
  discover → rank → write → gate flow. Returns `None` when the
  cadence floor hasn't elapsed or an open batch already exists for
  the niche.
* `await svc.show_batch(batch_id) -> BatchView` — unified ranked view
  of every candidate (external + internal merged), sorted by
  `effective_score = score * decay_factor` desc.
* `await svc.rank_batch(batch_id, ordered_candidate_ids)` — set
  `operator_rank` 1..N by position. Probes both candidate tables.
* `await svc.edit_winner(batch_id, topic=None, angle=None)` — rewrite
  the operator-edited topic/angle on the rank-1 candidate. Raises
  `ValueError` if nothing is ranked yet.
* `await svc.resolve_batch(batch_id)` — hand winner to the pipeline,
  flip batch status to `resolved`, record provenance.
* `await svc.reject_batch(batch_id, reason="")` — flip status to
  `expired` so the next sweep can claim the niche slot.
* Dataclasses: `BatchSnapshot`, `BatchView`, `CandidateView`.

## Configuration

All from `app_settings` via `site_config` (seeded in the baseline schema; originally migration 0119):

* `niche_top_n_per_pool` (default `5`) — top-N per pool fed into LLM
  final-score AND the slice inside `_embed_and_pre_rank`. Both ends
  of the funnel must stay consistent.
* `niche_carry_forward_decay_factor` (default `0.7`) — multiplier
  applied to each unpicked candidate's `decay_factor` per batch
  (0.7^3 ≈ 0.343 by batch 3).
* `niche_internal_rag_per_kind_limit` (default `4`) — per-kind cap on
  the internal RAG fetch (claude\_session, brain\_knowledge, etc).
* `niche_batch_expires_days` (default `7`) — `topic_batches.expires_at`.
* `topic_sanity_min_alpha_words` (default `2`) — minimum alphabetic
  words (letter-runs of ≥2 chars) for a topic to pass the sanity
  checkpoints; empty / zero-letter topics are always rejected.

Per-niche settings come from the `niches` table (set via NicheService),
not `app_settings`:

* `niche.batch_size` — final winner count.
* `niche.discovery_cadence_minute_floor` — min minutes between sweeps.
* `niche.writer_rag_mode` — threaded into `pipeline_tasks` on resolve.

## Dependencies

* **Reads from:**
  * `niches`, `niche_sources`, `niche_goals` via `NicheService`.
  * `topic_batches`, `topic_candidates`, `internal_topic_candidates`,
    `discovery_runs` for cadence + carry-forward + state.
  * `services.topic_ranking.embed_text`, `goal_vector_for`,
    `weighted_cosine_score`, `apply_decay`, `llm_final_score` (lazy
    imports — see module docstring on the test-seam rationale).
  * `services.internal_rag_source.InternalRagSource` for internal
    candidates.
* **Writes to:**
  * `discovery_runs` (start + finish/error).
  * `topic_batches` (open → resolved/expired with
    `picked_candidate_id` + `picked_candidate_kind` provenance).
  * `topic_candidates` and `internal_topic_candidates` (per-candidate
    rows with `score`, `score_breakdown`, `rank_in_batch`,
    `decay_factor`, `operator_rank`, `operator_edited_topic`,
    `operator_edited_angle`).
  * `pipeline_tasks` on `resolve_batch` — inserts a `pending` row with
    `topic_batch_id` set so provenance traces back from post to batch.
* **External APIs:** none directly. External topic sources (when
  wired) live in `services.topic_sources/` and call out from there.

## Failure modes

* **Contentless winner on resolve** — `_handoff_to_pipeline` emits a
  `topic_sanity_rejected` finding (severity `warn`) and raises
  `TopicSanityError` (a `ValueError`, so the HTTP route answers 400 and
  the CLI prints it). Operator paths: `edit_winner` to a real topic,
  then re-resolve. The `topic_auto_resolve` job catches this error and
  **expires the batch** instead of retry-failing every cycle — an open
  batch would otherwise block new sweeps for the niche.
* **Unregistered / failing external source** — `_discover_external`
  dispatches every enabled non-`internal_rag` TopicSource plugin for
  the niche; an unregistered source name or a per-source extract error
  is logged and skipped so one bad source can't starve the sweep.
* **Cadence floor blocks sweep** — `_floor_elapsed` returns `False`,
  `run_sweep` logs "Sweep skipped — discovery cadence floor not
  elapsed" and returns `None`. Even errored runs count toward the
  floor (intentional — don't hammer external sources when something's
  wedged).
* **Open batch already exists** — `_open_batch_exists` short-circuits.
  Resolve or reject the existing batch first.
* **No rank-1 on resolve** — `resolve_batch` raises
  `ValueError("no operator_rank=1 candidate; rank first")`. Operator
  must call `rank_batch` first.
* **Discovery error** — caught, written to `discovery_runs.error`,
  re-raised. The `topic_batches` row is rolled back by the absence of
  the INSERT (it's the LAST step in the try block).
* **`topic_decision` gate notifies, it doesn't block** — the durable gate
  state is `topic_batches.status = 'open'`, already written by
  `_write_batch` before `_open_topic_decision_gate` runs.
  `services.approval_service.pause_at_gate` doesn't apply here (it's
  `pipeline_tasks`-specific); `_open_topic_decision_gate` instead calls
  `notify_operator(..., critical=False)` (Discord, routine not critical)
  so the operator learns a batch is waiting (poindexter#862).

## Common ops

* **Trigger a sweep manually:**
  `await TopicBatchService(pool).run_sweep(niche_id=<uuid>)`.
* **List open batches per niche:**
  `SELECT n.slug, b.id, b.candidate_count, b.expires_at FROM topic_batches b JOIN niches n ON n.id = b.niche_id WHERE b.status = 'open';`
* **Inspect a batch:** `await svc.show_batch(batch_id=<uuid>)` returns
  a unified ranked list. UI-friendly.
* **Reject a stale open batch** to unblock the niche:
  `await svc.reject_batch(batch_id=<uuid>, reason="stale")`.
* **Tune carry-forward aggressiveness:** raise
  `niche_carry_forward_decay_factor` toward 1.0 to keep older
  candidates competitive longer; lower toward 0 to flush them faster.
* **Audit provenance for a published post:**
  `SELECT topic_batch_id FROM pipeline_tasks WHERE task_id = '<uuid>';`
  then look up the batch + winner via `picked_candidate_id`.

## See also

* `docs/architecture/niches-and-rag-modes.md` — niche/source/goal
  modeling and how `writer_rag_mode` flows downstream.
* `docs/architecture/services/content_router_service.md` — what
  happens after `resolve_batch` hands off the winner.