curated_feed.state

Reading and writing everything the pipeline keeps on disk.

Two kinds of file. The corpus is one JSONL per collection day, appended to and never rewritten. The artifacts are one JSON per stage per day, under the state directory, each the output of one stage and the input of the next.

That every stage boundary is a file is what makes –date work: a saved candidate set can be re-scored, or a saved response re-selected, without refetching anything. Policy tuning is then a loop of seconds rather than a day.

Attributes

CORPUS_SUFFIX
ARTIFACT_SUFFIX

Exceptions

StateError

Raised when a file the pipeline depends on is missing or unreadable.

Functions

corpus_path(→ pathlib.Path)	The corpus file for one collection day.
corpus_day(→ datetime.date)	The collection day a corpus file is named for.
latest_corpus_path(→ pathlib.Path)	The most recent corpus file, by date-ordered filename.
resolve_corpus(→ tuple[pathlib.Path, datetime.date])	The corpus file to work on, defaulting to the most recent collection day.
read_jsonl(→ list[RecordT])	Read and validate a JSONL file, skipping blank lines.
append_jsonl(→ int)	Append records as JSONL, creating the file and its directory if needed.
read_corpus(→ list[curated_feed.models.Candidate])	Read and validate a corpus file.
append_corpus(→ int)	Append candidates to a corpus file.
existing_ids(→ set[str])	Every id already recorded in the corpus directory.
artifact_path(→ pathlib.Path)	Where one stage's output for one day lives.
read_artifact(→ ArtifactT)	Read and validate one stage artifact.
write_artifact(→ pathlib.Path)	Write one stage artifact, indented so its daily diff stays readable.
write_text_artifact(→ pathlib.Path)	Write a stage artifact that is not a model, such as a raw response.

Module Contents

curated_feed.state.CORPUS_SUFFIX = '.jsonl'

curated_feed.state.ARTIFACT_SUFFIX = '.json'

exception curated_feed.state.StateError

Bases: Exception

Raised when a file the pipeline depends on is missing or unreadable.

curated_feed.state.corpus_path(corpus_dir: pathlib.Path, day: datetime.date) → pathlib.Path

The corpus file for one collection day.

>>> corpus_path(Path("/corpus"), date(2026, 8, 8)).as_posix()
'/corpus/2026-08-08.jsonl'

curated_feed.state.corpus_day(path: pathlib.Path) → datetime.date

The collection day a corpus file is named for.

>>> corpus_day(Path("/corpus/2026-08-08.jsonl")).isoformat()
'2026-08-08'

curated_feed.state.latest_corpus_path(corpus_dir: pathlib.Path) → pathlib.Path

The most recent corpus file, by date-ordered filename.

curated_feed.state.resolve_corpus(corpus_dir: pathlib.Path, day: datetime.date | None) → tuple[pathlib.Path, datetime.date]

The corpus file to work on, defaulting to the most recent collection day.

curated_feed.state.read_jsonl[RecordT: pydantic.BaseModel](path: pathlib.Path, model: type[RecordT]) → list[RecordT]

Read and validate a JSONL file, skipping blank lines.

A missing file reads as empty: the published log does not exist until the first feed is rendered, and that is not an error.

curated_feed.state.append_jsonl(path: pathlib.Path, records: collections.abc.Iterable[pydantic.BaseModel]) → int

Append records as JSONL, creating the file and its directory if needed.

Writes nothing when there is nothing to write: an empty corpus file would otherwise become the “latest” file that later commands pick up.

curated_feed.state.read_corpus(path: pathlib.Path) → list[curated_feed.models.Candidate]

Read and validate a corpus file.

curated_feed.state.append_corpus(path: pathlib.Path, candidates: collections.abc.Iterable[curated_feed.models.Candidate]) → int

Append candidates to a corpus file.

curated_feed.state.existing_ids(corpus_dir: pathlib.Path) → set[str]

Every id already recorded in the corpus directory.

Reads the id field alone rather than validating whole records: this runs over every line of every day collected so far, and only identity matters here. Junk lines are skipped, because a corrupt tail must not block a fetch.

curated_feed.state.artifact_path(state_dir: pathlib.Path, stage: str, day: datetime.date) → pathlib.Path

Where one stage’s output for one day lives.

>>> artifact_path(Path("/state"), "response", date(2026, 8, 8)).as_posix()
'/state/response/2026-08-08.json'

curated_feed.state.read_artifact[ArtifactT: pydantic.BaseModel](path: pathlib.Path, model: type[ArtifactT]) → ArtifactT

Read and validate one stage artifact.

curated_feed.state.write_artifact(path: pathlib.Path, artifact: pydantic.BaseModel) → pathlib.Path

Write one stage artifact, indented so its daily diff stays readable.

curated_feed.state.write_text_artifact(path: pathlib.Path, text: str) → pathlib.Path

Write a stage artifact that is not a model, such as a raw response.

The raw text is saved before parsing, so a response that fails validation can still be read afterwards.