From 4d2c491bcb9795913010ff3fc3fc7ecb6d6c045b Mon Sep 17 00:00:00 2001 From: sartaj Date: Thu, 16 Jul 2026 02:25:03 +0000 Subject: [PATCH] Make Govbot a DAG: manifest-driven pipeline with classify as a transform MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reshape govbot around a data-driven DAG: a govbot.yml manifest declares datasets, transforms, publishers, and named pipelines, and `govbot run` walks a pipeline by spawning each stage as a subprocess over the stream protocol (newline-JSON, stable `id`, typed `kind`). The built-in tagger is decoupled into an ordinary transform node: `govbot source --select docs | govbot classify | govbot apply`. Swapping the classifier (e.g. to `fastclass classify -`) is a one-line manifest `command` change — no code change. classify scores the `docs` projection's text via new `TagMatcher::match_text` / `match_tags_keywords_text` and is offline-first (keyword fallback, never downloads mid-pipeline). apply is the classification sink; both produce the same `.tag.json` format. Details: - config.rs: additive `Manifest` (uniform `Transform { command, reads, writes }` — no classify-specific fields; `Publisher` map; `pipelines`). Legacy `repos:`/`tags:` still parse; unknown keys ignored. 4 unit tests. - schemas/govbot.schema.json: DAG schema (datasets/transforms/publish/ pipelines); `required` relaxed for back-compat. - main.rs: `source --select docs` projection; new `classify`/`apply`/`run` commands; subcommand rename Clone/Logs/Build -> Pull/Source/Publish with old names kept as aliases; removed the deprecated `tag` command (split into classify+apply). Bare `govbot` runs the DAG when govbot.yml exists, wizard/init otherwise. - pipeline.rs: data-driven runner (linear walk of a DAG-capable manifest). - wizard.rs: `govbot init` scaffolds a DAG manifest (datasets + classify transform + default pipeline). - publish.rs: `get_repos_from_config` reads `datasets:` then `repos:`. - action.yml/justfile: migrated to the DAG (source|classify|apply, publish). Publisher stages currently emit via the existing `build` logic (minimal wiring); a per-publisher module split is a follow-up. Includes a crate `cargo fmt` normalization pass. 20 tests pass; verified end-to-end on the wy mock (`govbot run` produces tag files + feed.xml). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_014MAr5v6uw35YENW4HCCxyQ --- actions/govbot/action.yml | 24 +- actions/govbot/justfile | 34 +- actions/govbot/src/config.rs | 226 ++- actions/govbot/src/embeddings.rs | 29 +- actions/govbot/src/main.rs | 1543 +++++++++-------- actions/govbot/src/pipeline.rs | 253 +-- actions/govbot/src/publish.rs | 19 +- actions/govbot/src/wizard.rs | 62 +- actions/govbot/tests/api_snaps.rs | 15 +- ...i_example_snaps__snapshot@govbot_help.snap | 19 +- .../wizard_tests__wizard_all_no_tag.snap | 13 +- .../wizard_tests__wizard_all_with_tag.snap | 13 +- ...rd_tests__wizard_session_all_own_tags.snap | 13 +- ...rd_tests__wizard_session_all_with_tag.snap | 13 +- ...rd_tests__wizard_session_single_state.snap | 13 +- ...sts__wizard_session_specific_own_tags.snap | 13 +- ...sts__wizard_session_specific_with_tag.snap | 13 +- .../wizard_tests__wizard_single_with_tag.snap | 13 +- .../wizard_tests__wizard_specific_no_tag.snap | 13 +- actions/govbot/tests/wizard_tests.rs | 63 +- schemas/govbot.schema.json | 157 +- 21 files changed, 1550 insertions(+), 1011 deletions(-) diff --git a/actions/govbot/action.yml b/actions/govbot/action.yml index 1bac60aa..69e55c73 100644 --- a/actions/govbot/action.yml +++ b/actions/govbot/action.yml @@ -83,18 +83,18 @@ runs: exit 1 fi - # If cache was hit, govbot clone will just update existing repos (git pull) - # If cache was missed, govbot clone will do fresh clones - # When no repos are specified, govbot clone updates existing repos only + # If cache was hit, govbot pull will just update existing repos (git pull) + # If cache was missed, govbot pull will do fresh clones + # When no repos are specified, govbot pull updates existing repos only if [ "${{ steps.cache-repos.outputs.cache-hit }}" == "true" ]; then echo "📥 Cache hit - updating existing repositories..." # Update existing repos (no args = update existing only) - ${{ github.action_path }}/bin/govbot clone \ + ${{ github.action_path }}/bin/govbot pull \ --govbot-dir "$GOVBOT_DIR" || true else echo "📥 Cache miss - cloning all repositories..." # Clone all repos - ${{ github.action_path }}/bin/govbot clone all \ + ${{ github.action_path }}/bin/govbot pull all \ --govbot-dir "$GOVBOT_DIR" || true fi @@ -105,12 +105,16 @@ runs: path: ${{ steps.set-govbot-dir.outputs.repos-dir }} key: govbot-repos-${{ runner.os }}-${{ hashFiles('govbot.yml') }} - - name: Tag bills + - name: Classify bills shell: bash working-directory: ${{ github.workspace }} run: | - echo "🏷️ Tagging bills..." - ${{ github.action_path }}/bin/govbot logs | ${{ github.action_path }}/bin/govbot tag || true + echo "🏷️ Classifying bills..." + BIN="${{ github.action_path }}/bin/govbot" + # The classify stage of the DAG: source -> classify -> apply. + # Swap `govbot classify` for any stream-protocol classifier (e.g. + # `fastclass classify -`) via the `transforms:` block in govbot.yml. + "$BIN" source --select docs | "$BIN" classify | "$BIN" apply || true - name: Generate RSS feed id: publish @@ -147,9 +151,9 @@ runs: ARGS="$ARGS --govbot-dir ${{ inputs.govbot-dir }}" fi - # Run build command + # Run publish command # govbot.yml is automatically found in workspace root - ${{ github.action_path }}/bin/govbot build $ARGS + ${{ github.action_path }}/bin/govbot publish $ARGS # Determine output path (read from govbot.yml or use defaults) if [ -n "${{ inputs.output-dir }}" ]; then diff --git a/actions/govbot/justfile b/actions/govbot/justfile index 54ed620e..c23fc4cc 100644 --- a/actions/govbot/justfile +++ b/actions/govbot/justfile @@ -56,9 +56,9 @@ build-release: # Usage: just govbot [COMMAND] [ARGS...] # Examples: # just govbot --help -# just govbot clone usa il -# just govbot clone --govbot-dir custom-dir usa -# just govbot logs --repos usa +# just govbot pull usa il +# just govbot pull --govbot-dir custom-dir usa +# just govbot source --repos usa govbot *ARGS: #!/usr/bin/env bash set -e @@ -140,29 +140,11 @@ run: run-args ARGS: cargo run -- {{ARGS}} -# Tag bills using AI (reads JSON lines from stdin) -# Usage: just govbot logs --repos il --limit 10 | just govbot tag --ai-tool "ollama run llama3" -# Example: just govbot logs --repos il --limit 10 | just govbot tag --ai-tool "ollama" -# Note: The tag command reads from stdin, so pipe the logs output to it -tag *ARGS: - #!/usr/bin/env bash - set -e - - DEV_DIR=".govbot" - - # Build release binary if it doesn't exist or if any source files are newer - if [ ! -f "target/release/govbot" ] || [ "src" -nt "target/release/govbot" ] || find src -name "*.rs" -newer "target/release/govbot" 2>/dev/null | grep -q .; then - echo "🔨 Building release target..." - cargo build --release - fi - - # Check if --govbot-dir is already in the arguments - ARGS_STR="{{ARGS}}" - if [[ "$ARGS_STR" =~ --govbot-dir ]]; then - ./target/release/govbot tag {{ARGS}} - else - ./target/release/govbot tag {{ARGS}} --govbot-dir "$DEV_DIR" - fi +# Classification is now a DAG transform: run the pipeline with the generic +# `govbot` recipe, e.g.: +# just govbot source --select docs | just govbot classify | just govbot apply +# or run the whole manifest pipeline at once: +# just govbot run # Run the release binary run-release: diff --git a/actions/govbot/src/config.rs b/actions/govbot/src/config.rs index dfbf0d4e..4d8170c7 100644 --- a/actions/govbot/src/config.rs +++ b/actions/govbot/src/config.rs @@ -1,5 +1,155 @@ use crate::error::{Error, Result}; -use std::path::PathBuf; +use serde::Deserialize; +use std::collections::BTreeMap; +use std::path::{Path, PathBuf}; + +/// A `govbot.yml` project manifest describing a DAG of stages. +/// +/// govbot.yml declares the datasets a project consumes, the transforms it runs +/// over them, the publishers that emit artifacts, and named pipelines that wire +/// those stages together. `govbot run ` walks a pipeline's stages. +/// +/// The manifest is deliberately **not** a classifier: a transform node is a +/// uniform `{ command, reads, writes }`. The built-in tagger fills the classify +/// role as an ordinary transform whose `command` is `govbot classify`; swapping +/// in an external classifier (e.g. `fastclass classify -`) is only a `command` +/// change. There is no classify-specific field. +/// +/// Parsing is **additive**: the legacy `repos:` and `tags:` keys still parse so +/// pre-DAG projects and the built-in `govbot classify` node keep working. Unknown +/// keys (e.g. a legacy `build:` block) are ignored rather than rejected. +#[derive(Debug, Clone, Default, Deserialize)] +pub struct Manifest { + #[serde(default, rename = "$schema")] + pub schema: Option, + + /// Datasets the project consumes. Additive superset of the legacy `repos:`. + #[serde(default)] + pub datasets: Vec, + + /// Legacy dataset list; still honored so old manifests keep working. + #[serde(default)] + pub repos: Vec, + + /// Named external-process transform nodes. Uniform shape — no privileged + /// classify node; `govbot classify` and `fastclass classify -` are peers. + #[serde(default)] + pub transforms: BTreeMap, + + /// Named publisher nodes. Each consumes the result stream and emits one + /// artifact (an RSS feed or an HTML index in this release). + #[serde(default)] + pub publish: BTreeMap, + + /// Named `govbot run` targets: each an ordered list of stage names that + /// reference entries in `transforms` and `publish`. + #[serde(default)] + pub pipelines: BTreeMap>, +} + +impl Manifest { + /// Load and parse a `govbot.yml` manifest from disk. + pub fn load(path: impl AsRef) -> Result { + let contents = std::fs::read_to_string(path.as_ref()).map_err(|e| { + Error::Config(format!( + "Failed to read manifest {}: {}", + path.as_ref().display(), + e + )) + })?; + serde_yaml::from_str(&contents) + .map_err(|e| Error::Config(format!("Failed to parse govbot.yml: {}", e))) + } + + /// The dataset list, preferring `datasets:` and falling back to `repos:`. + pub fn dataset_list(&self) -> &[String] { + if !self.datasets.is_empty() { + &self.datasets + } else { + &self.repos + } + } +} + +/// A single external-process transform stage (a DAG node). +/// +/// A transform is a separate program that speaks the govbot stream protocol +/// (newline-delimited JSON on stdio, stable `id`, typed `kind`). govbot streams +/// records of the transform's `reads` kind into it and routes the records of its +/// `writes` kind back by `id`. +#[derive(Debug, Clone, Deserialize)] +pub struct Transform { + /// The stage command: a shell string (`"govbot classify"`) or an argv array + /// (`["govbot", "classify"]`). + pub command: CommandSpec, + /// The stream record kind this transform consumes (e.g. `docs`). + pub reads: String, + /// The stream record kind this transform produces (e.g. `classification`). + pub writes: String, +} + +/// A stage command: either a shell string or an explicit argv array. +#[derive(Debug, Clone, Deserialize)] +#[serde(untagged)] +pub enum CommandSpec { + /// `command: govbot classify` — split on whitespace. + Shell(String), + /// `command: ["govbot", "classify"]` — used verbatim. + Argv(Vec), +} + +impl CommandSpec { + /// The command split into program + arguments. + pub fn argv(&self) -> Vec { + match self { + CommandSpec::Shell(s) => s.split_whitespace().map(|s| s.to_string()).collect(), + CommandSpec::Argv(v) => v.clone(), + } + } +} + +/// The kind of artifact a publisher emits. Each kind emits exactly one artifact. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum PublisherKind { + /// Writes the RSS feed (default `feed.xml`). + Rss, + /// Writes the HTML index (default `index.html`). + Html, +} + +/// A single publisher node. +#[derive(Debug, Clone, Deserialize)] +pub struct Publisher { + #[serde(rename = "type")] + pub kind: PublisherKind, + + /// Tag names to include; if omitted, all tagged records are published. + #[serde(default)] + pub select: Option>, + + /// Base URL for generated links (e.g. the GitHub Pages URL). + #[serde(default)] + pub base_url: Option, + + /// Directory the publisher writes its artifact to. + #[serde(default)] + pub output_dir: Option, + + /// Output filename; defaults by kind (`rss` -> feed.xml, `html` -> index.html). + #[serde(default)] + pub output_file: Option, + + #[serde(default)] + pub title: Option, + + #[serde(default)] + pub description: Option, + + /// Max entries; a number, or the string `none` for all. + #[serde(default)] + pub limit: Option, +} /// Sort order for log entries #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -169,3 +319,77 @@ impl Default for Config { Self::new("tmp/repos") } } + +#[cfg(test)] +mod manifest_tests { + use super::*; + + const DAG_YML: &str = r#" +datasets: [wy, il] +tags: + clean_energy: + description: energy bills +transforms: + classify: + command: govbot classify + reads: docs + writes: classification + summarize: + command: ["fastclass", "summarize", "-"] + reads: docs + writes: summary +publish: + feed: + type: rss + base_url: https://example.org +pipelines: + default: [classify, feed] +build: + base_url: https://legacy.example +"#; + + #[test] + fn parses_dag_manifest() { + let m: Manifest = serde_yaml::from_str(DAG_YML).unwrap(); + assert_eq!(m.datasets, vec!["wy", "il"]); + assert_eq!(m.transforms.len(), 2); + assert_eq!(m.transforms["classify"].reads, "docs"); + assert_eq!(m.transforms["classify"].writes, "classification"); + assert_eq!(m.publish["feed"].kind, PublisherKind::Rss); + assert_eq!(m.pipelines["default"], vec!["classify", "feed"]); + } + + #[test] + fn command_spec_shell_and_argv_normalize() { + let m: Manifest = serde_yaml::from_str(DAG_YML).unwrap(); + // A shell string splits on whitespace… + assert_eq!( + m.transforms["classify"].command.argv(), + vec!["govbot", "classify"] + ); + // …and an explicit argv array is used verbatim. + assert_eq!( + m.transforms["summarize"].command.argv(), + vec!["fastclass", "summarize", "-"] + ); + } + + #[test] + fn legacy_repos_only_manifest_still_parses() { + // A pre-DAG manifest (repos + tags, no transforms) must remain valid, + // and unknown keys like `build:` are ignored, not rejected. + let m: Manifest = serde_yaml::from_str( + "repos: [wy]\ntags:\n x:\n description: y\nbuild:\n base_url: z\n", + ) + .unwrap(); + assert!(m.datasets.is_empty()); + assert_eq!(m.dataset_list(), &["wy".to_string()]); + assert!(m.transforms.is_empty()); + } + + #[test] + fn dataset_list_prefers_datasets_over_repos() { + let m: Manifest = serde_yaml::from_str("datasets: [a]\nrepos: [b]\n").unwrap(); + assert_eq!(m.dataset_list(), &["a".to_string()]); + } +} diff --git a/actions/govbot/src/embeddings.rs b/actions/govbot/src/embeddings.rs index feb68ab2..49edb369 100644 --- a/actions/govbot/src/embeddings.rs +++ b/actions/govbot/src/embeddings.rs @@ -404,15 +404,25 @@ impl TagMatcher { &self, value: &serde_json::Value, ) -> anyhow::Result> { - let text = ocd_files_select_default(value); + self.match_text(&ocd_files_select_default(value)) + } + + /// Score already-extracted text against the loaded tags. + /// + /// This is the stream-native scoring entry point: a `docs` record carries + /// its text pre-projected, and the `govbot classify` transform scores that + /// text directly (as an external classifier like fastclass would). + /// `match_json_value` is the convenience wrapper that projects an OCD value + /// to text first. + pub fn match_text(&self, text: &str) -> anyhow::Result> { let mut embeddings = self.embeddings.lock().unwrap(); - let log_embedding = embeddings.embed(&text)?; + let log_embedding = embeddings.embed(text)?; let mut results = Vec::new(); for (name, tag_def) in &self.tags { let score_breakdown = self.calculate_composite_score( &log_embedding, - &text, + text, name, tag_def, &mut *embeddings, @@ -443,7 +453,18 @@ pub fn match_tags_keywords( tag_defs: &[TagDefinition], json_entry: &serde_json::Value, ) -> Vec<(String, ScoreBreakdown)> { - let text = ocd_files_select_default(json_entry); + match_tags_keywords_text(tag_defs, &ocd_files_select_default(json_entry)) +} + +/// Keyword-based fallback matcher scoring already-extracted text. +/// +/// The stream-native counterpart to [`match_tags_keywords`]: the `govbot +/// classify` transform passes a `docs` record's pre-projected `text` here when +/// the embedding model is unavailable. +pub fn match_tags_keywords_text( + tag_defs: &[TagDefinition], + text: &str, +) -> Vec<(String, ScoreBreakdown)> { let text_lower = text.to_lowercase(); let mut results = Vec::new(); diff --git a/actions/govbot/src/main.rs b/actions/govbot/src/main.rs index 0b59ab15..a96e94da 100644 --- a/actions/govbot/src/main.rs +++ b/actions/govbot/src/main.rs @@ -1,18 +1,21 @@ use clap::{Parser, Subcommand}; +use futures::stream; +use futures::StreamExt; use govbot::git; -use govbot::{TagMatcher, hash_text, TagFile, TagFileMetadata, BillTagResult}; -use govbot::selectors::ocd_files_select_default; -use govbot::publish::{load_config, get_repos_from_config, filter_by_tags, deduplicate_entries, sort_by_timestamp}; +use govbot::publish::{ + deduplicate_entries, filter_by_tags, get_repos_from_config, load_config, sort_by_timestamp, +}; use govbot::rss; -use futures::StreamExt; -use futures::stream; -use std::io::{self, Write, BufRead, BufReader}; -use std::path::PathBuf; -use serde_json; +use govbot::selectors::ocd_files_select_default; +use govbot::{hash_text, BillTagResult, TagFile, TagFileMetadata, TagMatcher}; use jwalk::WalkDir; +use serde::{Deserialize, Serialize}; +use serde_json; +use std::collections::HashMap; use std::fs; +use std::io::{self, BufRead, BufReader, Write}; +use std::path::{Path, PathBuf}; use std::process::Command as ProcessCommand; -use std::collections::HashMap; /// Write a line to stdout, gracefully handling broken pipe errors /// This is essential for piping to tools like yq, jq, etc. @@ -40,7 +43,7 @@ fn write_json_line(line: &str) -> io::Result<()> { #[derive(Debug, Clone)] struct CloneResult { locale: String, - result: String, // "cloned", "pulled", "no_updates", "failed" + result: String, // "cloned", "pulled", "no_updates", "failed" position: String, // "1/37" size: Option, local_size: Option, @@ -60,10 +63,11 @@ struct Args { #[derive(Subcommand, Debug)] enum Command { - /// Clone or pull data pipeline repositories (default: updates existing repos) - /// Clones if repository doesn't exist, pulls if it does - /// Use "govbot clone all" to clone all repos, or "govbot clone " for specific repos - Clone { + /// Pull data pipeline datasets (default: updates existing datasets) + /// Clones if a dataset doesn't exist, pulls if it does. + /// Use "govbot pull all" for all datasets, or "govbot pull " for specific ones. + #[command(visible_alias = "clone")] + Pull { /// Repository names to clone/pull (e.g., usa, il, ca, or "all" for all repos). If not specified, updates existing repos. #[arg(num_args = 0..)] repos: Vec, @@ -89,12 +93,17 @@ enum Command { list: bool, }, - /// Process and display pipeline log files - Logs { + /// Emit the activity stream (the DAG's `source` stage). + /// + /// Walks the cloned datasets and emits one newline-JSON record per log + /// event. `--select docs` emits the stream-protocol `{id, text, kind:"docs", + /// sources}` projection that transforms (e.g. `govbot classify`) consume. + #[command(visible_alias = "logs")] + Source { /// Repos to output (default: `all`) `--repos="il,ca"` #[arg(long, num_args = 0..)] repos: Vec, - + /// Per repo limit (default: 100) options: `none` | number #[arg(long, default_value = "100")] limit: String, @@ -103,8 +112,9 @@ enum Command { #[arg(long, default_value = "bill,tags")] join: String, - /// Select/transform fields (default: `default`) - applies extract_text_from_json transformation - #[arg(long, default_value = "default", value_parser = ["default"])] + /// Select/transform fields (default: `default`) options: `default` | `docs`. + /// `docs` emits the stream-protocol projection `{id, text, kind:"docs", sources}`. + #[arg(long, default_value = "default", value_parser = ["default", "docs"])] select: String, /// Filter log entries based on per-repo AI generated filters (default: `default`) options: `default` | `none` @@ -117,7 +127,7 @@ enum Command { /// Govbot directory (default: $CWD/.govbot/repos, or GOVBOT_DIR env var) #[arg(long = "govbot-dir")] - govbot_dir: Option, + govbot_dir: Option, }, /// Delete data pipeline repositories @@ -165,54 +175,81 @@ enum Command { /// Downloads and installs the latest nightly build from GitHub releases Update, - /// Build RSS feed and HTML index from govbot.yml configuration - /// Generates a combined RSS feed and HTML index from logs filtered by tags in govbot.yml - Build { + /// Publish RSS feed and HTML index from govbot.yml (the DAG's publish stage) + /// Generates a combined RSS feed and HTML index from the stream filtered by tags. + #[command(visible_alias = "build")] + Publish { /// Specific tags to include in feed (default: all tags from govbot.yml) #[arg(long, num_args = 0..)] tags: Vec, - + /// Limit number of entries per feed (default: 100, use "none" for all entries) #[arg(long)] limit: Option, - + /// Output directory for RSS feed and HTML (default: from govbot.yml build.output_dir, or "docs") #[arg(long)] output_dir: Option, - + /// Output filename for RSS feed (default: from govbot.yml build.output_file, or "feed.xml") #[arg(long)] output_file: Option, - + /// Govbot directory (default: $CWD/.govbot/repos, or GOVBOT_DIR env var) #[arg(long = "govbot-dir")] govbot_dir: Option, }, - /// Tag bills using semantic or built-in similarity based on govbot.yml in the current directory. - /// Reads JSON lines from stdin (from `govbot logs`), processes entries with bill identifiers, - /// and writes per-tag files under the directory containing govbot.yml. - /// By default, acts as a filter: only outputs lines that match tags. - /// If a tag name is provided, only processes and outputs lines matching that specific tag. - Tag { - /// Optional tag name to filter to a specific tag (e.g., "lgbtq", "budget") - tag_name: Option, + /// Classify records from stdin with the built-in tagger (a DAG transform). + /// + /// Reads the `docs` stream (or any record carrying `text` / an OCD body) on + /// stdin, scores each against the taxonomy in `govbot.yml` (`tags:`), and + /// writes one `classification` record per input to stdout. This is the + /// built-in classify transform; an external `fastclass classify -` is an + /// interchangeable peer on the same stream contract. + Classify { + /// Path to the classifier config providing the tag taxonomy + /// (default: ./govbot.yml). + #[arg(long)] + classifier: Option, + /// Model/tokenizer directory (default: ./.govbot, or GOVBOT_DIR env var) + #[arg(long = "govbot-dir")] + govbot_dir: Option, + }, + + /// Apply classification records from stdin, persisting per-tag .tag.json files. + /// + /// The sink end of the classify pipeline: reads `classification` records + /// (from `govbot classify`, or any transform speaking the stream protocol) + /// and writes/merges + /// `/country:.../sessions//tags/.tag.json`. + Apply { /// Output directory (defaults to the directory containing govbot.yml) #[arg(long = "output-dir")] output_dir: Option, - /// Govbot directory (default: $CWD/.govbot/repos, or GOVBOT_DIR env var) + /// Govbot directory (default: $CWD/.govbot, or GOVBOT_DIR env var) #[arg(long = "govbot-dir")] govbot_dir: Option, + }, - /// Force re-tagging even if bill already exists in tag files + /// Run a manifest pipeline: source | transforms | apply, then publishers. + /// + /// Walks the named pipeline (or the sole/first pipeline) declared in + /// govbot.yml, executing each `transforms` entry as a subprocess over the + /// stream protocol, then running the `publish` stages. Every stage is an + /// opaque command — `govbot classify` and `fastclass classify -` are peers. + Run { + /// Pipeline name to run (default: the sole/first pipeline in govbot.yml) + pipeline: Option, + + /// Render publishers without emitting (propagated to publish stages) #[arg(long)] - overwrite: bool, + dry_run: bool, }, } - fn get_govbot_dir(govbot_dir: Option) -> anyhow::Result { // Check flag first, then environment variable, then default if let Some(govbot_dir) = govbot_dir { @@ -236,13 +273,13 @@ fn process_single_locale( ) -> CloneResult { let repo_name = git::build_repo_name(locale); let target_dir = repos_dir.join(&repo_name); - + let local_size = if target_dir.exists() { git::get_directory_size(&target_dir).unwrap_or(0) } else { 0 }; - + match git::clone_or_pull_repo_quiet(locale, repos_dir, token_str, !verbose) { Ok(action) => { let final_size = if target_dir.exists() { @@ -250,7 +287,7 @@ fn process_single_locale( } else { 0 }; - + let result = match action { "clone" => "🆕", "pulled" => "⬇️", @@ -258,7 +295,7 @@ fn process_single_locale( "recloned" => "🔄", _ => "processed", }; - + let mut clone_result = CloneResult { locale: locale.to_string(), result: result.to_string(), @@ -268,14 +305,14 @@ fn process_single_locale( final_size: None, error: None, }; - + if action == "clone" || action == "recloned" || action == "no_updates" { clone_result.size = Some(git::format_size(final_size)); } else { clone_result.local_size = Some(git::format_size(local_size)); clone_result.final_size = Some(git::format_size(final_size)); } - + clone_result } Err(e) => CloneResult { @@ -302,15 +339,17 @@ fn print_result(result: &CloneResult) { } else { let size_str = if let Some(ref size) = result.size { size.clone() - } else if let (Some(ref local), Some(ref final_size)) = (&result.local_size, &result.final_size) { + } else if let (Some(ref local), Some(ref final_size)) = + (&result.local_size, &result.final_size) + { format!("{} -> {}", local, final_size) } else { String::new() }; - + // result.result now contains the emoji directly (🆕, ⬇️, ✅, 🔄) let action_emoji = &result.result; - + if !size_str.is_empty() { eprintln!("{} {:<6} [{}]", action_emoji, result.locale, size_str); } else { @@ -331,7 +370,7 @@ async fn perform_clone_operations( ) -> anyhow::Result> { let total = repos_to_clone.len(); let mut all_results = Vec::new(); - + if total == 1 || num_jobs == 1 { // Sequential clone/pull - print as we go for (idx, locale) in repos_to_clone.iter().enumerate() { @@ -344,7 +383,7 @@ async fn perform_clone_operations( // Parallel clone/pull - print as results come in use std::sync::{Arc, Mutex}; let completed = Arc::new(Mutex::new(0usize)); - + let clone_futures = stream::iter(repos_to_clone.iter()) .map(|locale| { let locale = locale.clone(); @@ -353,9 +392,10 @@ async fn perform_clone_operations( let completed = completed.clone(); let total = total; let verbose_flag = verbose; - + tokio::task::spawn_blocking(move || { - let mut result = process_single_locale(&locale, &repos_dir, token.as_deref(), verbose_flag); + let mut result = + process_single_locale(&locale, &repos_dir, token.as_deref(), verbose_flag); let mut count = completed.lock().unwrap(); *count += 1; result.position = format!("{}/{}", *count, total); @@ -365,7 +405,7 @@ async fn perform_clone_operations( .buffer_unordered(num_jobs); let mut stream = clone_futures; - + while let Some(result) = stream.next().await { match result { Ok(data) => { @@ -391,20 +431,20 @@ async fn perform_clone_operations( let _ = std::io::stderr().flush(); } } - + Ok(all_results) } - async fn run_clone_command(cmd: Command) -> anyhow::Result<()> { - let Command::Clone { + let Command::Pull { repos, govbot_dir, token, parallel, verbose, list, - } = cmd else { + } = cmd + else { unreachable!() }; @@ -420,19 +460,23 @@ async fn run_clone_command(cmd: Command) -> anyhow::Result<()> { } let repos_dir = get_govbot_dir(govbot_dir)?; - + // Get token from argument or environment variable let env_token = std::env::var("TOKEN").ok(); let token_str = token.as_deref().or(env_token.as_deref()); - + // Get parallelization setting let num_jobs = parallel - .or_else(|| std::env::var("GOVBOT_JOBS").ok().and_then(|s| s.parse().ok())) + .or_else(|| { + std::env::var("GOVBOT_JOBS") + .ok() + .and_then(|s| s.parse().ok()) + }) .unwrap_or(4); // Parse repos and handle "all" let mut repos_to_clone = Vec::new(); - + if repos.is_empty() { // No repos specified: find existing repos to update // Check all known locales to see which repos exist @@ -441,32 +485,32 @@ async fn run_clone_command(cmd: Command) -> anyhow::Result<()> { let locale_str = locale.as_lowercase(); let repo_name = git::build_repo_name(&locale_str); let repo_path = repos_dir.join(&repo_name); - + // Check if this is a git repository if repo_path.exists() && repo_path.join(".git").exists() { repos_to_clone.push(locale_str.to_string()); } } - + if repos_to_clone.is_empty() { eprintln!("No repos downloaded yet in this directory"); eprintln!("to download all gov data, do `govbot clone all`. future syncs are just `govbot clone`"); return Ok(()); } - + // Create directory if it doesn't exist (needed for the clone operations) std::fs::create_dir_all(&repos_dir)?; } else { // Create directory if it doesn't exist (needed for the clone operations) std::fs::create_dir_all(&repos_dir)?; - + // Parse specified repos for repo in repos { let repo = repo.trim().to_lowercase(); if repo.is_empty() { continue; } - + if repo == "all" { // Add all working locales let all_locales = govbot::locale::WorkingLocale::all(); @@ -483,42 +527,35 @@ async fn run_clone_command(cmd: Command) -> anyhow::Result<()> { if repos_to_clone.is_empty() { return Ok(()); -} + } // Print initial message with count eprintln!("🔁 Syncing {} repos\n", repos_to_clone.len()); // Perform clone operations and print results as they complete - let results = perform_clone_operations( - repos_to_clone, - repos_dir, - token_str, - num_jobs, - verbose, - ).await?; - + let results = + perform_clone_operations(repos_to_clone, repos_dir, token_str, num_jobs, verbose).await?; + // Show summary - let errors: Vec<_> = results.iter() - .filter(|r| r.result == "failed") - .collect(); - + let errors: Vec<_> = results.iter().filter(|r| r.result == "failed").collect(); + if !errors.is_empty() { eprintln!("\n❌ Errors occurred: {}/{}", errors.len(), results.len()); } else if !results.is_empty() { eprintln!("\n✅ Successfully processed all {} repos!", results.len()); } - + Ok(()) } - async fn run_delete_command(cmd: Command) -> anyhow::Result<()> { let Command::Delete { locales, govbot_dir, parallel, verbose, - } = cmd else { + } = cmd + else { unreachable!() }; @@ -537,10 +574,14 @@ async fn run_delete_command(cmd: Command) -> anyhow::Result<()> { } let repos_dir = get_govbot_dir(govbot_dir)?; - + // Get parallelization setting let num_jobs = parallel - .or_else(|| std::env::var("GOVBOT_JOBS").ok().and_then(|s| s.parse().ok())) + .or_else(|| { + std::env::var("GOVBOT_JOBS") + .ok() + .and_then(|s| s.parse().ok()) + }) .unwrap_or(4); // Parse locales and handle "all" @@ -550,7 +591,7 @@ async fn run_delete_command(cmd: Command) -> anyhow::Result<()> { if locale.is_empty() { continue; } - + if locale == "all" { // Add all working locales let all_locales = govbot::locale::WorkingLocale::all(); @@ -575,18 +616,18 @@ async fn run_delete_command(cmd: Command) -> anyhow::Result<()> { let total = locales_to_delete.len(); let mut deleted_count = 0; let mut failed_count = 0; - + if total == 1 || num_jobs == 1 { // Sequential delete for (idx, locale) in locales_to_delete.iter().enumerate() { let repo_name = format!("{}-data-pipeline", locale); let target_dir = repos_dir.join(&repo_name); let existed = target_dir.exists(); - + if verbose { eprintln!("[{}/{}] Deleting {}...", idx + 1, total, locale); } - + match git::delete_repo(locale, &repos_dir) { Ok(_) => { if existed { @@ -607,7 +648,7 @@ async fn run_delete_command(cmd: Command) -> anyhow::Result<()> { use std::sync::{Arc, Mutex}; let deleted = Arc::new(Mutex::new(0usize)); let failed = Arc::new(Mutex::new(0usize)); - + let delete_futures = stream::iter(locales_to_delete.iter()) .map(|locale| { let locale = locale.clone(); @@ -616,18 +657,18 @@ async fn run_delete_command(cmd: Command) -> anyhow::Result<()> { let failed = failed.clone(); let total = total; let verbose_flag = verbose; - + tokio::task::spawn_blocking(move || { let repo_name = format!("{}-data-pipeline", locale); let target_dir = repos_dir.join(&repo_name); - + if verbose_flag { let d = deleted.lock().unwrap(); let f = failed.lock().unwrap(); let current = *d + *f + 1; eprintln!("[{}/{}] Deleting {}...", current, total, locale); } - + let existed = target_dir.exists(); match git::delete_repo(&locale, &repos_dir) { Ok(_) => { @@ -650,7 +691,7 @@ async fn run_delete_command(cmd: Command) -> anyhow::Result<()> { .buffer_unordered(num_jobs); let mut stream = delete_futures; - + while let Some(result) = stream.next().await { match result { Ok((locale, Ok(status))) => { @@ -666,11 +707,11 @@ async fn run_delete_command(cmd: Command) -> anyhow::Result<()> { } } } - + deleted_count = *deleted.lock().unwrap(); failed_count = *failed.lock().unwrap(); } - + // Show summary if failed_count > 0 { eprintln!("\n❌ Errors occurred: {}/{}", failed_count, total); @@ -679,12 +720,12 @@ async fn run_delete_command(cmd: Command) -> anyhow::Result<()> { } else { eprintln!("\n✅ No repositories found to delete."); } - + Ok(()) } async fn run_logs_command(cmd: Command) -> anyhow::Result<()> { - let Command::Logs { + let Command::Source { govbot_dir, repos, sort: _sort, @@ -692,10 +733,11 @@ async fn run_logs_command(cmd: Command) -> anyhow::Result<()> { join, select, filter, - } = cmd else { + } = cmd + else { unreachable!() }; - + // Parse join options - now supports field paths like "bill.title" and special "tags" let mut join_specs: Vec<(String, Vec)> = Vec::new(); let mut join_tags = false; @@ -715,7 +757,11 @@ async fn run_logs_command(cmd: Command) -> anyhow::Result<()> { let limit_parsed: Option = if limit.to_lowercase() == "none" { None } else { - Some(limit.parse().map_err(|e| anyhow::anyhow!("Invalid limit value '{}': {}", limit, e))?) + Some( + limit + .parse() + .map_err(|e| anyhow::anyhow!("Invalid limit value '{}': {}", limit, e))?, + ) }; // Parse comma-separated repos if provided as single string @@ -741,7 +787,7 @@ async fn run_logs_command(cmd: Command) -> anyhow::Result<()> { if locale.is_empty() { continue; } - + if locale == "all" { // Find all existing repos in the directory if git_dir.exists() { @@ -750,7 +796,7 @@ async fn run_logs_command(cmd: Command) -> anyhow::Result<()> { let locale_str = loc.as_lowercase(); let repo_name = git::build_repo_name(&locale_str); let repo_path = git_dir.join(&repo_name); - + // Only add repos that actually exist (for logs, we don't need .git, just the directory) if repo_path.exists() && repo_path.is_dir() { repos_to_process.push(repo_name); @@ -772,7 +818,7 @@ async fn run_logs_command(cmd: Command) -> anyhow::Result<()> { // Process each repo (with optional filtering) for repo_name in repos_to_process { let repo_path = git_dir.join(&repo_name); - + if !repo_path.exists() { eprintln!("Warning: Repository not found: {}", repo_path.display()); continue; @@ -781,7 +827,7 @@ async fn run_logs_command(cmd: Command) -> anyhow::Result<()> { // Walk the repo directory to find log files matching the pattern: // repo_name/country:{country}/state:{state}/sessions/{session_name}/logs/*.json let mut file_count = 0; - + for entry_result in WalkDir::new(&repo_path) .process_read_dir(|_depth, _path, _read_dir_state, _children| { // Optional: customize directory reading behavior @@ -801,7 +847,7 @@ async fn run_logs_command(cmd: Command) -> anyhow::Result<()> { } let path = entry.path(); - + // Check if it's a JSON file in a logs directory if !path.is_file() { continue; @@ -814,7 +860,7 @@ async fn run_logs_command(cmd: Command) -> anyhow::Result<()> { // Check if path matches: country:{country}/state:{state}/sessions/{session_name}/logs/*.json let path_str = path.to_string_lossy(); let repo_prefix = repo_path.to_string_lossy(); - + // Get relative path by stripping the repo prefix // Handle both absolute and relative paths let relative_path = if let Some(stripped) = path_str.strip_prefix(&*repo_prefix) { @@ -824,11 +870,11 @@ async fn run_logs_command(cmd: Command) -> anyhow::Result<()> { // If prefix doesn't match, skip this file continue; }; - + // Match pattern: country:*/state:*/sessions/*/logs/*.json // Use a simple regex-like check: must have these components in order - if relative_path.starts_with("country:") - && relative_path.contains("/state:") + if relative_path.starts_with("country:") + && relative_path.contains("/state:") && relative_path.contains("/sessions/") && relative_path.contains("/logs/") && relative_path.ends_with(".json") @@ -838,12 +884,12 @@ async fn run_logs_command(cmd: Command) -> anyhow::Result<()> { let state_pos = relative_path.find("/state:").unwrap_or(usize::MAX); let sessions_pos = relative_path.find("/sessions/").unwrap_or(usize::MAX); let logs_pos = relative_path.find("/logs/").unwrap_or(usize::MAX); - + // Verify order: country < state < sessions < logs if country_pos < state_pos && state_pos < sessions_pos && sessions_pos < logs_pos { // Compute relative source path let source_path_str = compute_relative_source_path(&path, &git_dir); - + // Read JSON file, parse it, and build extensible output structure match fs::read_to_string(&path) { Ok(contents) => { @@ -857,19 +903,22 @@ async fn run_logs_command(cmd: Command) -> anyhow::Result<()> { .or_else(|| json_value.get("bill_identifier")) .and_then(|id| id.as_str()) .map(|s| s.to_string()); - + // Build output with extensible structure: // - Data keys (log, bill, etc.) are singular entity names matching source keys // - sources object automatically tracks all data sources let mut output = serde_json::Map::new(); - + // Add the log data with key "log" (matching sources.log) output.insert("log".to_string(), json_value); - + // Add sources with the log path let mut sources = serde_json::Map::new(); - sources.insert("log".to_string(), serde_json::Value::String(source_path_str.clone())); - + sources.insert( + "log".to_string(), + serde_json::Value::String(source_path_str.clone()), + ); + // Join additional datasets if requested for (dataset_name, field_path) in &join_specs { match dataset_name.as_str() { @@ -881,36 +930,59 @@ async fn run_logs_command(cmd: Command) -> anyhow::Result<()> { Ok(p) => p, Err(_) => path.clone(), }; - - let metadata_path = canonical_log_path.parent() + + let metadata_path = canonical_log_path + .parent() .and_then(|logs_dir| { logs_dir.parent().map(|bill_dir| { bill_dir.join("metadata.json") }) }); - + if let Some(ref metadata_path) = metadata_path { if metadata_path.exists() { match fs::read_to_string(metadata_path) { Ok(metadata_contents) => { - match serde_json::from_str::(&metadata_contents) { + match serde_json::from_str::< + serde_json::Value, + >( + &metadata_contents + ) { Ok(metadata_value) => { // If field_path is specified, extract just that field // Otherwise, include the full bill data if field_path.is_empty() { // No field path specified, include full bill data - output.insert("bill".to_string(), metadata_value); + output.insert( + "bill".to_string(), + metadata_value, + ); } else { // Extract specific field(s) from bill data - if let Some(field_value) = extract_json_field(&metadata_value, field_path) { + if let Some( + field_value, + ) = + extract_json_field( + &metadata_value, + field_path, + ) + { // Use the full join path as the key (e.g., "bill.title") - let output_key = format!("{}.{}", dataset_name, field_path.join(".")); - output.insert(output_key, field_value); + let output_key = format!( + "{}.{}", + dataset_name, + field_path + .join(".") + ); + output.insert( + output_key, + field_value, + ); } else { eprintln!("Warning: Field path {:?} not found in metadata from {}", field_path, metadata_path.display()); } } - + // Add bill source path let bill_source_path = compute_relative_source_path(metadata_path, &git_dir); sources.insert("bill".to_string(), serde_json::Value::String(bill_source_path)); @@ -932,38 +1004,54 @@ async fn run_logs_command(cmd: Command) -> anyhow::Result<()> { } } _ => { - eprintln!("Warning: Unknown join dataset: {}", dataset_name); + eprintln!( + "Warning: Unknown join dataset: {}", + dataset_name + ); } } } - + // Join tags if requested if join_tags { // Extract country, state, session_id from the path - if let Some((country, state, session_id)) = extract_path_info(&source_path_str) { + if let Some((country, state, session_id)) = + extract_path_info(&source_path_str) + { // Use bill_id extracted earlier if let Some(ref bill_id) = bill_id_opt { // Look for tags in cwd/country:us/state:{state}/sessions/{session_id}/tags/ - let cwd = std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")); + let cwd = std::env::current_dir() + .unwrap_or_else(|_| PathBuf::from(".")); let tags_dir = cwd .join(&format!("country:{}", country)) .join(&format!("state:{}", state)) .join("sessions") .join(&session_id) .join("tags"); - + if tags_dir.exists() && tags_dir.is_dir() { let mut matched_tags = serde_json::Map::new(); if let Ok(entries) = fs::read_dir(&tags_dir) { for entry in entries.flatten() { let path = entry.path(); // Check for both .tag.json and .json files - if let Some(ext) = path.extension().and_then(|s| s.to_str()) { + if let Some(ext) = path + .extension() + .and_then(|s| s.to_str()) + { if ext == "json" { - if let Some(stem) = path.file_stem().and_then(|s| s.to_str()) { + if let Some(stem) = path + .file_stem() + .and_then(|s| s.to_str()) + { // Remove .tag suffix if present (e.g., "budget.tag" -> "budget") - let tag_name = stem.strip_suffix(".tag").unwrap_or(stem); - match fs::read_to_string(&path) { + let tag_name = stem + .strip_suffix(".tag") + .unwrap_or(stem); + match fs::read_to_string( + &path, + ) { Ok(contents) => { if let Ok(tag_file) = serde_json::from_str::(&contents) { // Check if bill_id exists in bills map @@ -981,48 +1069,75 @@ async fn run_logs_command(cmd: Command) -> anyhow::Result<()> { } } if !matched_tags.is_empty() { - output.insert("tags".to_string(), serde_json::Value::Object(matched_tags)); + output.insert( + "tags".to_string(), + serde_json::Value::Object(matched_tags), + ); } } } } } - - output.insert("sources".to_string(), serde_json::Value::Object(sources)); - + + output.insert( + "sources".to_string(), + serde_json::Value::Object(sources), + ); + // Extract timestamp from sources.log path (after "logs/" and before "_") // Do this after sources is inserted so we can use the final sources.log value let timestamp = extract_timestamp_from_path(&source_path_str); if let Some(ref ts) = timestamp { - output.insert("timestamp".to_string(), serde_json::Value::String(ts.clone())); + output.insert( + "timestamp".to_string(), + serde_json::Value::String(ts.clone()), + ); } - + let mut output_value = serde_json::Value::Object(output); - + // Apply select transformation if requested if select == "default" { // Select specific keys from nested objects, preserving structure let mut selected_output = serde_json::Map::new(); - + // Top: id (from log.bill_id), then log object with selected fields - if let Some(id) = output_value.get("log").and_then(|l| l.get("bill_id").or_else(|| l.get("bill_identifier"))).and_then(|v| v.as_str()) { - selected_output.insert("id".to_string(), serde_json::Value::String(id.to_string())); + if let Some(id) = output_value + .get("log") + .and_then(|l| { + l.get("bill_id") + .or_else(|| l.get("bill_identifier")) + }) + .and_then(|v| v.as_str()) + { + selected_output.insert( + "id".to_string(), + serde_json::Value::String(id.to_string()), + ); } - + // Create log object with only action and bill_id if let Some(log) = output_value.get("log") { let mut log_obj = serde_json::Map::new(); if let Some(action) = log.get("action") { - log_obj.insert("action".to_string(), action.clone()); + log_obj + .insert("action".to_string(), action.clone()); } - if let Some(bill_id) = log.get("bill_id").or_else(|| log.get("bill_identifier")) { - log_obj.insert("bill_id".to_string(), bill_id.clone()); + if let Some(bill_id) = log + .get("bill_id") + .or_else(|| log.get("bill_identifier")) + { + log_obj + .insert("bill_id".to_string(), bill_id.clone()); } if !log_obj.is_empty() { - selected_output.insert("log".to_string(), serde_json::Value::Object(log_obj)); + selected_output.insert( + "log".to_string(), + serde_json::Value::Object(log_obj), + ); } } - + // Create bill object with only selected fields if let Some(bill) = output_value.get("bill") { let mut bill_obj = serde_json::Map::new(); @@ -1030,54 +1145,111 @@ async fn run_logs_command(cmd: Command) -> anyhow::Result<()> { bill_obj.insert("title".to_string(), title.clone()); } if let Some(abstracts) = bill.get("abstracts") { - bill_obj.insert("abstracts".to_string(), abstracts.clone()); + bill_obj.insert( + "abstracts".to_string(), + abstracts.clone(), + ); } if let Some(subject) = bill.get("subject") { - bill_obj.insert("subject".to_string(), subject.clone()); + bill_obj + .insert("subject".to_string(), subject.clone()); } if let Some(identifier) = bill.get("identifier") { - bill_obj.insert("identifier".to_string(), identifier.clone()); + bill_obj.insert( + "identifier".to_string(), + identifier.clone(), + ); } if let Some(session) = bill.get("legislative_session") { - bill_obj.insert("legislative_session".to_string(), session.clone()); + bill_obj.insert( + "legislative_session".to_string(), + session.clone(), + ); } if let Some(org) = bill.get("from_organization") { - bill_obj.insert("from_organization".to_string(), org.clone()); + bill_obj.insert( + "from_organization".to_string(), + org.clone(), + ); } if !bill_obj.is_empty() { - selected_output.insert("bill".to_string(), serde_json::Value::Object(bill_obj)); + selected_output.insert( + "bill".to_string(), + serde_json::Value::Object(bill_obj), + ); } } - + // Always include tags (even if empty/null) since it's part of the default selector if let Some(tags) = output_value.get("tags") { - selected_output.insert("tags".to_string(), tags.clone()); + selected_output + .insert("tags".to_string(), tags.clone()); } else { // Include empty tags object if not present - selected_output.insert("tags".to_string(), serde_json::Value::Null); + selected_output.insert( + "tags".to_string(), + serde_json::Value::Null, + ); } - + // Bottom: sources, timestamp if let Some(sources) = output_value.get("sources") { - selected_output.insert("sources".to_string(), sources.clone()); + selected_output + .insert("sources".to_string(), sources.clone()); } if let Some(timestamp) = output_value.get("timestamp") { - selected_output.insert("timestamp".to_string(), timestamp.clone()); + selected_output + .insert("timestamp".to_string(), timestamp.clone()); } - + output_value = serde_json::Value::Object(selected_output); } - + // Apply filter - let should_output = match filter_manager.should_keep(&output_value, &repo_name) { + let should_output = match filter_manager + .should_keep(&output_value, &repo_name) + { govbot::FilterResult::Keep => true, govbot::FilterResult::FilterOut => false, }; - + if should_output { + // `--select docs`: emit the stream-protocol projection + // `{id, text, kind:"docs", sources}`. Done after filtering + // so the per-repo filter still sees `log.action`. + if select == "docs" { + let id = output_value + .get("log") + .and_then(|l| { + l.get("bill_id") + .or_else(|| l.get("bill_identifier")) + }) + .and_then(|v| v.as_str()) + .map(|s| s.to_string()) + .unwrap_or_default(); + let text = ocd_files_select_default(&output_value); + let mut doc = serde_json::Map::new(); + doc.insert( + "id".to_string(), + serde_json::Value::String(id), + ); + doc.insert( + "kind".to_string(), + serde_json::Value::String("docs".to_string()), + ); + doc.insert( + "text".to_string(), + serde_json::Value::String(text), + ); + if let Some(sources) = output_value.get("sources") { + doc.insert("sources".to_string(), sources.clone()); + } + output_value = serde_json::Value::Object(doc); + } + // Deep prune empty/null values before serialization let pruned_value = deep_prune_json(output_value); - + // Serialize as compact JSON (single line) match serde_json::to_string(&pruned_value) { Ok(json_line) => { @@ -1087,7 +1259,11 @@ async fn run_logs_command(cmd: Command) -> anyhow::Result<()> { } } Err(e) => { - eprintln!("Error serializing JSON from {}: {}", path.display(), e); + eprintln!( + "Error serializing JSON from {}: {}", + path.display(), + e + ); } } } @@ -1109,28 +1285,30 @@ async fn run_logs_command(cmd: Command) -> anyhow::Result<()> { Ok(()) } - /// Parse a join string like "bill.title" into (dataset_name, field_path) fn parse_join_string(join_str: &str) -> Option<(String, Vec)> { let parts: Vec<&str> = join_str.split('.').collect(); if parts.is_empty() { return None; } - + let dataset_name = parts[0].to_string(); let field_path = if parts.len() > 1 { parts[1..].iter().map(|s| s.to_string()).collect() } else { Vec::new() }; - + Some((dataset_name, field_path)) } /// Extract a value from JSON using a field path (e.g., ["title"] or ["bill", "title"]) -fn extract_json_field(value: &serde_json::Value, field_path: &[String]) -> Option { +fn extract_json_field( + value: &serde_json::Value, + field_path: &[String], +) -> Option { let mut current = value; - + for field in field_path { match current { serde_json::Value::Object(map) => { @@ -1146,7 +1324,7 @@ fn extract_json_field(value: &serde_json::Value, field_path: &[String]) -> Optio _ => return None, } } - + Some(current.clone()) } @@ -1220,13 +1398,13 @@ fn compute_relative_source_path(file_path: &PathBuf, git_dir: &PathBuf) -> Strin Ok(p) => p, Err(_) => file_path.clone(), }; - + // Canonicalize git_dir for proper relative path calculation let canonical_git_dir = match git_dir.canonicalize() { Ok(p) => p, Err(_) => git_dir.clone(), }; - + // Get relative path from git_dir to the file match pathdiff::diff_paths(&canonical_file, &canonical_git_dir) { Some(rel_path) => rel_path.to_string_lossy().replace('\\', "/"), @@ -1245,7 +1423,8 @@ async fn run_load_command(cmd: Command) -> anyhow::Result<()> { govbot_dir, memory_limit, threads, - } = cmd else { + } = cmd + else { unreachable!() }; @@ -1253,23 +1432,25 @@ async fn run_load_command(cmd: Command) -> anyhow::Result<()> { // Check if directory exists if !repos_dir.exists() { - eprintln!("Error: Govbot repos directory not found: {}", repos_dir.display()); + eprintln!( + "Error: Govbot repos directory not found: {}", + repos_dir.display() + ); eprintln!("Run 'govbot clone all' first to clone repositories."); return Ok(()); } // Get base govbot directory (parent of repos) // e.g., if repos_dir is ./.govbot/repos, base_dir is ./.govbot - let base_govbot_dir = repos_dir.parent() + let base_govbot_dir = repos_dir + .parent() .ok_or_else(|| anyhow::anyhow!("Could not determine base govbot directory"))?; - + // Ensure base directory exists std::fs::create_dir_all(base_govbot_dir)?; // Check if duckdb is available - let duckdb_check = ProcessCommand::new("duckdb") - .arg("--version") - .output(); + let duckdb_check = ProcessCommand::new("duckdb").arg("--version").output(); if duckdb_check.is_err() { eprintln!("Error: 'duckdb' command not found."); @@ -1279,7 +1460,8 @@ async fn run_load_command(cmd: Command) -> anyhow::Result<()> { // Database file goes in the base govbot directory // Resolve to absolute path to ensure it's created in the right location - let db_path = base_govbot_dir.canonicalize() + let db_path = base_govbot_dir + .canonicalize() .unwrap_or_else(|_| base_govbot_dir.to_path_buf()) .join(&database); let db_path_str = db_path.to_string_lossy().to_string(); @@ -1322,7 +1504,10 @@ async fn run_load_command(cmd: Command) -> anyhow::Result<()> { sql_script.push_str("SELECT \n"); sql_script.push_str(" *,\n"); sql_script.push_str(" filename as source_file\n"); - sql_script.push_str(&format!("FROM read_json_auto('{}/**/bills/*/metadata.json', \n", repos_dir_str)); + sql_script.push_str(&format!( + "FROM read_json_auto('{}/**/bills/*/metadata.json', \n", + repos_dir_str + )); sql_script.push_str(" filename=true, \n"); sql_script.push_str(" union_by_name=true);\n"); sql_script.push_str("\n"); @@ -1354,7 +1539,7 @@ async fn run_load_command(cmd: Command) -> anyhow::Result<()> { duckdb_cmd.stderr(std::process::Stdio::piped()); let mut child = duckdb_cmd.spawn()?; - + // Write SQL to stdin if let Some(mut stdin) = child.stdin.take() { stdin.write_all(sql_script.as_bytes())?; @@ -1393,164 +1578,191 @@ async fn run_load_command(cmd: Command) -> anyhow::Result<()> { fn extract_path_info(path: &str) -> Option<(String, String, String)> { // Find country: pattern let country_start = path.find("country:")?; - let country_end = path[country_start + 8..].find('/').unwrap_or(path.len() - country_start - 8); + let country_end = path[country_start + 8..] + .find('/') + .unwrap_or(path.len() - country_start - 8); let country = path[country_start + 8..country_start + 8 + country_end].to_string(); - + // Find state: pattern let state_start = path.find("/state:")?; - let state_end = path[state_start + 7..].find('/').unwrap_or(path.len() - state_start - 7); + let state_end = path[state_start + 7..] + .find('/') + .unwrap_or(path.len() - state_start - 7); let state = path[state_start + 7..state_start + 7 + state_end].to_string(); - + // Find sessions/ pattern let sessions_start = path.find("/sessions/")?; - let session_end = path[sessions_start + 10..].find('/').unwrap_or(path.len() - sessions_start - 10); + let session_end = path[sessions_start + 10..] + .find('/') + .unwrap_or(path.len() - sessions_start - 10); let session_id = path[sessions_start + 10..sessions_start + 10 + session_end].to_string(); - + Some((country, state, session_id)) } -/// Download a file from a URL to a local path -fn download_file(url: &str, path: &std::path::Path) -> anyhow::Result<()> { - eprintln!("Downloading {}...", url); - let response = reqwest::blocking::get(url)?; - if !response.status().is_success() { - return Err(anyhow::anyhow!("Failed to download {}: HTTP {}", url, response.status())); - } - let mut file = std::fs::File::create(path)?; - std::io::copy(&mut response.bytes()?.as_ref(), &mut file)?; - Ok(()) -} +/// Tag result structure: (tag_key, score_breakdown) +type TagResult = (String, govbot::ScoreBreakdown); -/// Ensure embedding model and tokenizer exist; if missing, download them from Hugging Face. -/// Returns true if files are present/ready, false otherwise. -fn ensure_embedding_files(model_dir: &std::path::Path) -> bool { - let model_path = model_dir.join("model.onnx"); - let tokenizer_path = model_dir.join("tokenizer.json"); - let _vocab_path = model_dir.join("vocab.txt"); +/// A `classification` stream record — the output of a classify transform and +/// the input to `govbot apply`. Carries enough to persist tag files without +/// re-reading the source: the bill `id`, its `text`, the `sources` routing +/// block, and the matched `tags` (name -> score). This is the shape an external +/// classifier (e.g. `fastclass classify -`) emits too, so `apply` is agnostic +/// to which transform produced it. +#[derive(Debug, Serialize, Deserialize)] +struct ClassificationRecord { + id: String, + #[serde(default)] + kind: String, + #[serde(default)] + text: String, + #[serde(default)] + sources: serde_json::Value, + tags: HashMap, +} - if model_path.exists() && tokenizer_path.exists() { - return true; +/// Persist scored tags for one bill into per-tag `.tag.json` files. +/// +/// The shared sink used by `govbot apply`: routes via `sources.log` to +/// `/country:.../state:.../sessions//tags/.tag.json`, creating or +/// merging each tag file. Split out of the legacy in-process tagger so the same +/// on-disk format is produced whether classification came from the built-in +/// `govbot classify` or an external transform. +fn write_tag_files( + base_output_dir: &Path, + sources: &serde_json::Value, + bill_id: &str, + bill_text: &str, + tags: Vec, + tag_defs: &[govbot::TagDefinition], + model_str: &str, +) -> anyhow::Result<()> { + if tags.is_empty() { + return Ok(()); } - eprintln!("Embedding files not found. Downloading all-MiniLM-L6-v2 (ONNX) to {}...", model_dir.display()); - - // Use Xenova ONNX exports - let onnx_url = "https://huggingface.co/Xenova/all-MiniLM-L6-v2/resolve/main/onnx/model.onnx"; - let tokenizer_url = "https://huggingface.co/Xenova/all-MiniLM-L6-v2/resolve/main/tokenizer.json"; + let (country, state, session_id) = sources + .get("log") + .and_then(|p| p.as_str()) + .and_then(extract_path_info) + .unwrap_or_else(|| { + ( + "us".to_string(), + "unknown".to_string(), + "unknown".to_string(), + ) + }); - // Download tokenizer.json - if !tokenizer_path.exists() { - if let Err(e) = download_file(tokenizer_url, &tokenizer_path) { - eprintln!("Failed to download tokenizer.json: {}", e); - return false; - } - } + let tags_dir = base_output_dir + .join(format!("country:{}", country)) + .join(format!("state:{}", state)) + .join("sessions") + .join(&session_id) + .join("tags"); + fs::create_dir_all(&tags_dir)?; + + let text_hash = hash_text(bill_text); + let now = chrono::Utc::now().to_rfc3339(); + + for (tag_key, score_breakdown) in tags { + let tag_path = tags_dir.join(format!("{}.tag.json", tag_key)); + + let tag_def = tag_defs + .iter() + .find(|td| td.name == tag_key) + .cloned() + .unwrap_or_else(|| govbot::TagDefinition { + name: tag_key.clone(), + description: String::new(), + examples: Vec::new(), + include_keywords: Vec::new(), + exclude_keywords: Vec::new(), + negative_examples: Vec::new(), + threshold: 0.5, + }); + let tag_config_hash = hash_text(&serde_json::to_string(&tag_def)?); + + let mut tag_file: TagFile = if tag_path.exists() { + fs::read_to_string(&tag_path) + .ok() + .and_then(|c| serde_json::from_str(&c).ok()) + .unwrap_or_else(|| TagFile { + metadata: TagFileMetadata { + last_run: now.clone(), + model: model_str.to_string(), + tag_config_hash: tag_config_hash.clone(), + }, + tag_config: tag_def.clone(), + text_cache: HashMap::new(), + bills: HashMap::new(), + }) + } else { + TagFile { + metadata: TagFileMetadata { + last_run: now.clone(), + model: model_str.to_string(), + tag_config_hash: tag_config_hash.clone(), + }, + tag_config: tag_def.clone(), + text_cache: HashMap::new(), + bills: HashMap::new(), + } + }; - // Download ONNX model - if !model_path.exists() { - if let Err(e) = download_file(onnx_url, &model_path) { - eprintln!("Failed to download ONNX model: {}", e); - return false; + // Refresh metadata; adopt the current tag config if it drifted. + tag_file.metadata.last_run = now.clone(); + tag_file.metadata.model = model_str.to_string(); + if tag_config_hash != tag_file.metadata.tag_config_hash { + tag_file.tag_config = tag_def; + tag_file.metadata.tag_config_hash = tag_config_hash; } - } - if !model_path.exists() || !tokenizer_path.exists() { - eprintln!( - "Download completed but model.onnx or tokenizer.json not found in {}", - model_dir.display() + tag_file + .text_cache + .entry(text_hash.clone()) + .or_insert_with(|| bill_text.to_string()); + + tag_file.bills.insert( + bill_id.to_string(), + BillTagResult { + text_hash: text_hash.clone(), + score: score_breakdown, + }, ); - return false; - } - - eprintln!("✅ Successfully downloaded embedding files!"); - true -} - -/// Tag result structure: (tag_key, score_breakdown) -type TagResult = (String, govbot::ScoreBreakdown); -/// Check if a bill is already tagged in tag file(s) for the given session -/// If tag_name is Some, only checks that specific tag file -/// Returns a list of tag names that contain this bill -fn check_existing_tags( - tags_dir: &PathBuf, - bill_id: &str, - tag_name: Option<&str>, -) -> anyhow::Result> { - let mut matched_tags = Vec::new(); - - if !tags_dir.exists() { - return Ok(matched_tags); - } - - // If a specific tag is requested, only check that tag file - if let Some(requested_tag) = tag_name { - let tag_path = tags_dir.join(format!("{}.tag.json", requested_tag)); - if tag_path.exists() { - match fs::read_to_string(&tag_path) { - Ok(contents) => { - if let Ok(tag_file) = serde_json::from_str::(&contents) { - if tag_file.bills.contains_key(bill_id) { - matched_tags.push(requested_tag.to_string()); - } - } - } - Err(_) => { - // Tag file exists but can't be read - return empty - } - } - } - return Ok(matched_tags); + fs::write(&tag_path, serde_json::to_string_pretty(&tag_file)?)?; } - - // Otherwise, scan all .tag.json files in the tags directory - for entry in fs::read_dir(tags_dir)? { - let entry = entry?; - let path = entry.path(); - - if let Some(ext) = path.extension() { - if ext == "json" { - if let Some(stem) = path.file_stem().and_then(|s| s.to_str()) { - // Remove .tag suffix if present (e.g., "budget.tag" -> "budget") - let tag_name = stem.strip_suffix(".tag").unwrap_or(stem); - - match fs::read_to_string(&path) { - Ok(contents) => { - if let Ok(tag_file) = serde_json::from_str::(&contents) { - // Check if bill_id exists in bills map - if tag_file.bills.contains_key(bill_id) { - matched_tags.push(tag_name.to_string()); - } - } - } - Err(_) => { - // Skip files that can't be read - continue; - } - } - } - } - } - } - - Ok(matched_tags) + + Ok(()) } -async fn run_tag_command(cmd: Command) -> anyhow::Result<()> { - let Command::Tag { - tag_name, - output_dir, +/// `govbot classify` — the built-in tagger as a DAG transform node. +/// +/// Reads records from stdin (a `docs` projection, or any record carrying `text` +/// / an OCD body), scores each against the `govbot.yml` taxonomy, and emits one +/// `classification` record per matched input to stdout. Interchangeable with an +/// external `fastclass classify -`: same stdin/stdout stream contract. +async fn run_classify_command(cmd: Command) -> anyhow::Result<()> { + let Command::Classify { + classifier, govbot_dir, - overwrite, - } = cmd else { + } = cmd + else { unreachable!() }; - // Check if govbot.yml exists in current directory let current_dir = std::env::current_dir()?; - let default_tags_cfg = current_dir.join("govbot.yml"); + let cfg_path = classifier + .map(PathBuf::from) + .unwrap_or_else(|| current_dir.join("govbot.yml")); + if !cfg_path.exists() { + return Err(anyhow::anyhow!( + "classifier config not found: {}", + cfg_path.display() + )); + } + let tag_defs = govbot::embeddings::load_tags_config(&cfg_path)?; - // Model/tokenizer directory: prefer user-specified govbot-dir or env GOVBOT_DIR, else default .govbot let model_dir: PathBuf = if let Some(ref dir) = govbot_dir { PathBuf::from(dir) } else if let Ok(dir) = std::env::var("GOVBOT_DIR") { @@ -1558,400 +1770,205 @@ async fn run_tag_command(cmd: Command) -> anyhow::Result<()> { } else { current_dir.join(".govbot") }; - fs::create_dir_all(&model_dir)?; let model_path = model_dir.join("model.onnx"); let tokenizer_path = model_dir.join("tokenizer.json"); - - // Require govbot.yml - if !default_tags_cfg.exists() { - return Err(anyhow::anyhow!( - "govbot.yml not found in current directory" - )); - } - - // Load tag definitions (needed for both embedding and keyword fallback) - let tag_defs = govbot::embeddings::load_tags_config(&default_tags_cfg) - .map_err(|e| anyhow::anyhow!("Failed to parse govbot.yml: {}", e))?; - // Try embedding mode first - let embedding_matcher = if ensure_embedding_files(&model_dir) { - let tags_path = default_tags_cfg.clone(); - - eprintln!("Using embedding mode:"); - eprintln!(" Model: {}", model_path.display()); - eprintln!(" Tokenizer: {}", tokenizer_path.display()); - eprintln!(" Tags config: {}", tags_path.display()); - - match TagMatcher::from_files(&model_path, &tokenizer_path, &tags_path) { - Ok(matcher) => Some(matcher), + // Offline-first: use the embedding model only if it is already present. + // Unlike the interactive tagger, a transform never blocks on a download — + // it falls back to keyword matching so `govbot run` works without network. + let matcher = if model_path.exists() && tokenizer_path.exists() { + match TagMatcher::from_files(&model_path, &tokenizer_path, &cfg_path) { + Ok(m) => { + eprintln!("govbot classify: embedding mode ({})", model_path.display()); + Some(m) + } Err(e) => { - eprintln!("Warning: Failed to initialize embedding matcher: {}", e); - eprintln!("Falling back to keyword-based matching."); + eprintln!("Warning: embedding matcher init failed ({e}); using keyword matching"); None } } } else { - eprintln!("Embedding files not available; using keyword-based matching."); - eprintln!(" Tags config: {}", default_tags_cfg.display()); + eprintln!( + "govbot classify: keyword-based matching (no embedding model in {})", + model_dir.display() + ); None }; - - // Determine output directory - // If govbot.yml exists, use its directory as the base output directory - let base_output_dir = if default_tags_cfg.exists() { - // Use the directory containing govbot.yml - default_tags_cfg.parent() - .unwrap_or(¤t_dir) - .to_path_buf() - } else if let Some(ref dir) = output_dir { + + let stdin = io::stdin(); + let reader = BufReader::new(stdin.lock()); + let mut emitted = 0usize; + for line in reader.lines() { + let line = line?; + let line = line.trim(); + if line.is_empty() { + continue; + } + let value: serde_json::Value = match serde_json::from_str(line) { + Ok(v) => v, + Err(_) => continue, + }; + let id = value + .get("id") + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(); + // Prefer an explicit `text` (the `docs` projection); else project the + // text from the OCD body so a raw `source` stream also classifies. + let text = value + .get("text") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()) + .unwrap_or_else(|| ocd_files_select_default(&value)); + + let scored: Vec = if let Some(m) = matcher.as_ref() { + match m.match_text(&text) { + Ok(r) => r, + Err(e) => { + eprintln!("scoring error for {id}: {e}; falling back to keywords"); + govbot::embeddings::match_tags_keywords_text(&tag_defs, &text) + } + } + } else { + govbot::embeddings::match_tags_keywords_text(&tag_defs, &text) + }; + + if scored.is_empty() { + continue; + } + + let record = ClassificationRecord { + id, + kind: "classification".to_string(), + text, + sources: value + .get("sources") + .cloned() + .unwrap_or(serde_json::Value::Null), + tags: scored.into_iter().collect(), + }; + write_json_line(&serde_json::to_string(&record)?)?; + emitted += 1; + } + eprintln!("govbot classify: emitted {emitted} classification records"); + Ok(()) +} + +/// `govbot apply` — the sink end of the classify pipeline. +/// +/// Reads `classification` records from stdin (from `govbot classify` or any +/// transform speaking the protocol) and persists per-tag `.tag.json` files via +/// [`write_tag_files`]. +async fn run_apply_command(cmd: Command) -> anyhow::Result<()> { + let Command::Apply { + output_dir, + govbot_dir, + } = cmd + else { + unreachable!() + }; + + let current_dir = std::env::current_dir()?; + let cfg_path = current_dir.join("govbot.yml"); + // Tag definitions are only used for the `tag_config` metadata block; an + // external classifier that carries its own taxonomy still applies fine. + let tag_defs = if cfg_path.exists() { + govbot::embeddings::load_tags_config(&cfg_path).unwrap_or_default() + } else { + Vec::new() + }; + + let base_output_dir = if let Some(ref dir) = output_dir { PathBuf::from(dir) + } else if cfg_path.exists() { + cfg_path.parent().unwrap_or(¤t_dir).to_path_buf() } else if let Some(ref dir) = govbot_dir { PathBuf::from(dir) - } else if let Ok(dir) = std::env::var("GOVBOT_DIR") { - PathBuf::from(dir) } else { - // Default to current directory - current_dir + current_dir.clone() }; - - // Read JSON lines from stdin + let stdin = io::stdin(); let reader = BufReader::new(stdin.lock()); - - let mut processed_count = 0; - let mut skipped_count = 0; - let mut read_count: usize = 0; - - eprintln!("Reading JSON lines from stdin..."); - - for line_result in reader.lines() { - let line = line_result?; + let mut applied = 0usize; + for line in reader.lines() { + let line = line?; let line = line.trim(); if line.is_empty() { - read_count += 1; - if read_count % 100 == 0 { - eprintln!("Read {} lines (processed {}, skipped {})...", read_count, processed_count, skipped_count); - } continue; } - - read_count += 1; - // Parse JSON line (assumes default selector format) - match serde_json::from_str::(line) { - Ok(json_value) => { - // Extract bill_id from top-level "id" field (default selector format) - let bill_id_opt = json_value - .get("id") - .and_then(|id| id.as_str()); - - // Extract text from JSON for embedding comparison - let bill_text = ocd_files_select_default(&json_value); - - // Extract path info from sources.log (default selector format) - let path_info = json_value - .get("sources") - .and_then(|sources| sources.get("log")) - .and_then(|path| path.as_str()) - .and_then(|log_path| extract_path_info(log_path)) - .or_else(|| { - // Fallback: use default values if we can't determine - Some(("us".to_string(), "unknown".to_string(), "unknown".to_string())) - }); - - // Process if we have path info (from sources.log in default selector format) - if let Some((country, state, session_id)) = path_info { - // Get bill_id - use "id" from default selector, or generate from text hash if missing - let bill_id = bill_id_opt.map(|s| s.to_string()).unwrap_or_else(|| { - let text_hash = hash_text(&bill_text); - format!("entry_{}", &text_hash[..8]) - }); - - // Determine tags directory - let tags_dir = base_output_dir - .join(&format!("country:{}", country)) - .join(&format!("state:{}", state)) - .join("sessions") - .join(&session_id) - .join("tags"); - - // Validate tag_name if provided - if let Some(ref requested_tag) = tag_name { - if !tag_defs.iter().any(|td| td.name == *requested_tag) { - return Err(anyhow::anyhow!( - "Tag '{}' not found in govbot.yml. Available tags: {}", - requested_tag, - tag_defs.iter().map(|td| td.name.clone()).collect::>().join(", ") - )); - } - } - - // Fast path: check if bill is already tagged (unless overwrite is set) - let mut matched_tags: Vec = Vec::new(); - let mut should_run_tagging = overwrite; - - if !overwrite { - match check_existing_tags(&tags_dir, &bill_id, tag_name.as_deref()) { - Ok(existing_tags) => { - if !existing_tags.is_empty() { - // Bill is already tagged - output the line and skip tagging - matched_tags = existing_tags; - should_run_tagging = false; - } else { - // Bill not found in tag file(s) - need to run tagging - should_run_tagging = true; - } - } - Err(e) => { - // Error checking tags - run tagging to be safe - eprintln!("Warning: Error checking existing tags for {}: {}", bill_id, e); - should_run_tagging = true; - } - } - } - - // Run tagging logic if needed - if should_run_tagging { - // Choose strategy based on mode - let mut tags: Vec = if let Some(matcher) = embedding_matcher.as_ref() { - match matcher.match_json_value(&json_value) { - Ok(results) => results, - Err(e) => { - eprintln!("Error running embedding matcher for bill {}: {}", bill_id, e); - eprintln!("Falling back to keyword-based matching for this entry."); - // Fall back to keyword matching for this entry - govbot::embeddings::match_tags_keywords(&tag_defs, &json_value) - } - } - } else { - // Use keyword-based fallback matcher - govbot::embeddings::match_tags_keywords(&tag_defs, &json_value) - }; - - // Filter to specific tag if requested - if let Some(ref requested_tag) = tag_name { - tags.retain(|(tag, _)| tag == requested_tag); - } - - // Extract tag names from results - matched_tags = tags.iter().map(|(tag_name, _)| tag_name.clone()).collect(); - - // Save tags to files if we found matches - if !tags.is_empty() { - let text_hash = hash_text(&bill_text); - - // Write per-tag files immediately - fs::create_dir_all(&tags_dir)?; - - // Get current timestamp for metadata - let now = chrono::Utc::now().to_rfc3339(); - let model_path_str = if embedding_matcher.is_some() { - model_path.to_string_lossy().to_string() - } else { - "keyword-fallback".to_string() - }; - - for (tag_key, score_breakdown) in tags { - let tag_path = tags_dir.join(format!("{}.tag.json", tag_key)); - - // Load or create TagFile structure - let mut tag_file: TagFile = if tag_path.exists() { - match fs::read_to_string(&tag_path) { - Ok(contents) => { - serde_json::from_str(&contents).unwrap_or_else(|_| { - // If parsing fails, create a new TagFile - let tag_def = tag_defs - .iter() - .find(|td| td.name == tag_key) - .cloned() - .unwrap_or_else(|| govbot::TagDefinition { - name: tag_key.clone(), - description: String::new(), - examples: Vec::new(), - include_keywords: Vec::new(), - exclude_keywords: Vec::new(), - negative_examples: Vec::new(), - threshold: 0.5, - }); - - let tag_config_hash = hash_text(&serde_json::to_string(&tag_def).unwrap_or_default()); - - TagFile { - metadata: TagFileMetadata { - last_run: now.clone(), - model: model_path_str.clone(), - tag_config_hash, - }, - tag_config: tag_def, - text_cache: HashMap::new(), - bills: HashMap::new(), - } - }) - } - Err(_) => { - // Create new TagFile - let tag_def = tag_defs - .iter() - .find(|td| td.name == tag_key) - .cloned() - .unwrap_or_else(|| govbot::TagDefinition { - name: tag_key.clone(), - description: String::new(), - examples: Vec::new(), - include_keywords: Vec::new(), - exclude_keywords: Vec::new(), - negative_examples: Vec::new(), - threshold: 0.5, - }); - - let tag_config_hash = hash_text(&serde_json::to_string(&tag_def)?); - - TagFile { - metadata: TagFileMetadata { - last_run: now.clone(), - model: model_path_str.clone(), - tag_config_hash, - }, - tag_config: tag_def, - text_cache: HashMap::new(), - bills: HashMap::new(), - } - } - } - } else { - // Create new TagFile - let tag_def = tag_defs - .iter() - .find(|td| td.name == tag_key) - .cloned() - .unwrap_or_else(|| govbot::TagDefinition { - name: tag_key.clone(), - description: String::new(), - examples: Vec::new(), - include_keywords: Vec::new(), - exclude_keywords: Vec::new(), - negative_examples: Vec::new(), - threshold: 0.5, - }); - - let tag_config_hash = hash_text(&serde_json::to_string(&tag_def)?); - - TagFile { - metadata: TagFileMetadata { - last_run: now.clone(), - model: model_path_str.clone(), - tag_config_hash, - }, - tag_config: tag_def, - text_cache: HashMap::new(), - bills: HashMap::new(), - } - }; - - // Update metadata - tag_file.metadata.last_run = now.clone(); - tag_file.metadata.model = model_path_str.clone(); - - // Update tag config if it changed - let current_tag_def = tag_defs - .iter() - .find(|td| td.name == tag_key) - .cloned() - .unwrap_or_else(|| tag_file.tag_config.clone()); - - let current_config_hash = hash_text(&serde_json::to_string(¤t_tag_def)?); - if current_config_hash != tag_file.metadata.tag_config_hash { - tag_file.tag_config = current_tag_def; - tag_file.metadata.tag_config_hash = current_config_hash; - } - - // Add text to cache if not present - if !tag_file.text_cache.contains_key(&text_hash) { - tag_file.text_cache.insert(text_hash.clone(), bill_text.clone()); - } - - // Add/update bill result - tag_file.bills.insert(bill_id.to_string(), BillTagResult { - text_hash: text_hash.clone(), - score: score_breakdown, - }); - - // Write updated TagFile - let json_string = serde_json::to_string_pretty(&tag_file)?; - fs::write(&tag_path, json_string)?; - } - } - } - - // Output the line if it matches tags (filter mode) - // If a specific tag was requested, only output if that tag matches - // Otherwise, output if any tag matches - let should_output = if let Some(ref requested_tag) = tag_name { - matched_tags.contains(requested_tag) - } else { - !matched_tags.is_empty() - }; - - if should_output { - write_json_line(line)?; - } - - processed_count += 1; - if processed_count % 50 == 0 { - eprintln!("Processed {} entries (matched: {} tags)...", processed_count, matched_tags.len()); - } - } else { - // No path info - skip this entry (default selector should always provide sources.log) - skipped_count += 1; - } - } - Err(_e) => { - // Skip malformed/empty lines quietly - skipped_count += 1; - } - } - - if read_count % 100 == 0 { - eprintln!("Read {} lines (processed {}, skipped {})...", read_count, processed_count, skipped_count); + let record: ClassificationRecord = match serde_json::from_str(line) { + Ok(r) => r, + Err(_) => continue, + }; + if record.tags.is_empty() { + continue; } + let tags: Vec = record.tags.into_iter().collect(); + write_tag_files( + &base_output_dir, + &record.sources, + &record.id, + &record.text, + tags, + &tag_defs, + "govbot-classify", + )?; + applied += 1; } - - eprintln!("\nProcessed: {}, Skipped: {}", processed_count, skipped_count); - eprintln!("\n✅ Tagging complete!"); - + eprintln!("✅ govbot apply: wrote tag files for {applied} bills"); Ok(()) } +/// `govbot run` — execute a manifest pipeline (the DAG). +fn run_run_command(cmd: Command) -> anyhow::Result<()> { + let Command::Run { pipeline, dry_run } = cmd else { + unreachable!() + }; + let cwd = std::env::current_dir()?; + let config_path = cwd.join("govbot.yml"); + if !config_path.exists() { + return Err(anyhow::anyhow!( + "govbot.yml not found in current directory. Run `govbot init` first." + )); + } + govbot::pipeline::run_manifest_pipeline(&config_path, pipeline.as_deref(), dry_run) +} + async fn run_build_command(cmd: Command) -> anyhow::Result<()> { - let Command::Build { + let Command::Publish { tags, limit, output_dir, output_file, govbot_dir, - } = cmd else { + } = cmd + else { unreachable!() }; - + // Check if govbot.yml exists in current directory let current_dir = std::env::current_dir()?; let config_path = current_dir.join("govbot.yml"); - + if !config_path.exists() { return Err(anyhow::anyhow!("govbot.yml not found in current directory")); } - + // Load configuration let config = load_config(&config_path)?; - + // Get tags configuration - let tags_config = config.get("tags") + let tags_config = config + .get("tags") .and_then(|t| t.as_object()) .ok_or_else(|| anyhow::anyhow!("No tags found in configuration"))?; - + // Determine which tags to use let tags_to_use: Vec = if tags.is_empty() { // Use tags from build config, or all tags - if let Some(build_tags) = config.get("build") + if let Some(build_tags) = config + .get("build") .and_then(|p| p.get("tags")) .and_then(|t| t.as_array()) { @@ -1965,21 +1982,21 @@ async fn run_build_command(cmd: Command) -> anyhow::Result<()> { } else { tags }; - + // Validate tags exist for tag in &tags_to_use { if !tags_config.contains_key(tag) { return Err(anyhow::anyhow!("Tag '{}' not found in configuration", tag)); } } - + if tags_to_use.is_empty() { return Err(anyhow::anyhow!("No valid tags to process")); } - + // Get build configuration let build_config = config.get("build").and_then(|p| p.as_object()); - + // Get output directory let output_dir_path = if let Some(dir) = output_dir { PathBuf::from(dir) @@ -1990,7 +2007,7 @@ async fn run_build_command(cmd: Command) -> anyhow::Result<()> { .unwrap_or("docs"); PathBuf::from(dir_str) }; - + // Get output filename let output_filename = if let Some(file) = output_file { file @@ -2001,28 +2018,34 @@ async fn run_build_command(cmd: Command) -> anyhow::Result<()> { .unwrap_or("feed.xml") .to_string() }; - + // Get feed metadata let feed_title = build_config .and_then(|p| p.get("title")) .and_then(|t| t.as_str()) .map(|s| s.to_string()) .unwrap_or_else(|| { - format!("{} Legislation", tags_to_use.iter() - .map(|t| t.replace('_', " ").split_whitespace() - .map(|w| { - let mut chars = w.chars(); - match chars.next() { - None => String::new(), - Some(f) => f.to_uppercase().collect::() + chars.as_str(), - } - }) + format!( + "{} Legislation", + tags_to_use + .iter() + .map(|t| t + .replace('_', " ") + .split_whitespace() + .map(|w| { + let mut chars = w.chars(); + match chars.next() { + None => String::new(), + Some(f) => f.to_uppercase().collect::() + chars.as_str(), + } + }) + .collect::>() + .join(" ")) .collect::>() - .join(" ")) - .collect::>() - .join(" & ")) + .join(" & ") + ) }); - + let feed_description = build_config .and_then(|p| p.get("description")) .and_then(|d| d.as_str()) @@ -2032,12 +2055,16 @@ async fn run_build_command(cmd: Command) -> anyhow::Result<()> { for tag_name in &tags_to_use { if let Some(tag_obj) = tags_config.get(tag_name).and_then(|t| t.as_object()) { if let Some(desc) = tag_obj.get("description").and_then(|d| d.as_str()) { - let tag_title = tag_name.replace('_', " ").split_whitespace() + let tag_title = tag_name + .replace('_', " ") + .split_whitespace() .map(|w| { let mut chars = w.chars(); match chars.next() { None => String::new(), - Some(f) => f.to_uppercase().collect::() + chars.as_str(), + Some(f) => { + f.to_uppercase().collect::() + chars.as_str() + } } }) .collect::>() @@ -2052,40 +2079,38 @@ async fn run_build_command(cmd: Command) -> anyhow::Result<()> { descs.join(" | ") } }); - + let feed_link = build_config .and_then(|p| p.get("base_url")) .and_then(|u| u.as_str()) .unwrap_or("https://example.com"); - + let base_url = Some(feed_link); - + // Get repos let repos = get_repos_from_config(&config); - + // Get repos to process let repos_to_process: Vec = if repos == vec!["all".to_string()] { Vec::new() // Empty means all repos } else { repos }; - + // Get limit - parse "none" as no limit, otherwise parse as usize // Default to 100 if not specified let limit_str_opt = limit.or_else(|| { - build_config - .and_then(|p| p.get("limit")) - .and_then(|l| { - if let Some(s) = l.as_str() { - Some(s.to_string()) - } else if let Some(n) = l.as_u64() { - Some(n.to_string()) - } else { - None - } - }) + build_config.and_then(|p| p.get("limit")).and_then(|l| { + if let Some(s) = l.as_str() { + Some(s.to_string()) + } else if let Some(n) = l.as_u64() { + Some(n.to_string()) + } else { + None + } + }) }); - + let limit_value: Option = if let Some(limit_str) = limit_str_opt { if limit_str.to_lowercase() == "none" { None // No limit @@ -2095,11 +2120,14 @@ async fn run_build_command(cmd: Command) -> anyhow::Result<()> { } else { Some(100) // Default to 100 items }; - + // Run logs command and collect entries - eprintln!("Collecting log entries for tags: {}", tags_to_use.join(", ")); + eprintln!( + "Collecting log entries for tags: {}", + tags_to_use.join(", ") + ); let mut entries = Vec::new(); - + // Get the base govbot directory (not the repos subdirectory) // The logs command expects the base directory and will append /repos itself let base_govbot_dir = if let Some(ref gd) = govbot_dir { @@ -2114,12 +2142,11 @@ async fn run_build_command(cmd: Command) -> anyhow::Result<()> { .to_string_lossy() .to_string() }; - + // Call logs command as subprocess and parse JSON output // Use current executable (govbot binary) - let exe = std::env::current_exe() - .unwrap_or_else(|_| PathBuf::from("govbot")); - + let exe = std::env::current_exe().unwrap_or_else(|_| PathBuf::from("govbot")); + let mut cmd = ProcessCommand::new(exe); cmd.arg("logs") .arg("--join") @@ -2130,32 +2157,35 @@ async fn run_build_command(cmd: Command) -> anyhow::Result<()> { .arg("default") .arg("--sort") .arg("DESC"); - + // Only add --govbot-dir if it's not the default if !base_govbot_dir.is_empty() && base_govbot_dir != ".govbot" { cmd.arg("--govbot-dir").arg(&base_govbot_dir); } - + if !repos_to_process.is_empty() { cmd.arg("--repos"); for repo in &repos_to_process { cmd.arg(repo); } } - + // Don't pass limit to logs command - we'll limit after filtering/sorting // This ensures we get the best entries, not just the first N from each repo - + let output = cmd.output()?; - + // Check return code if !output.status.success() { let stderr_str = String::from_utf8_lossy(&output.stderr); - eprintln!("Error: logs command failed with exit code: {:?}", output.status.code()); + eprintln!( + "Error: logs command failed with exit code: {:?}", + output.status.code() + ); eprintln!("Stderr: {}", stderr_str); return Err(anyhow::anyhow!("Failed to collect log entries")); } - + // Check if there were any errors in stderr (but compilation messages are OK) if !output.stderr.is_empty() { let stderr_str = String::from_utf8_lossy(&output.stderr); @@ -2168,16 +2198,16 @@ async fn run_build_command(cmd: Command) -> anyhow::Result<()> { eprintln!("Warning from logs command: {}", filtered_stderr.join("\n")); } } - + // Parse JSON lines from output let mut total_entries = 0; let mut filtered_entries = 0; let stdout_str = String::from_utf8_lossy(&output.stdout); - + if stdout_str.trim().is_empty() { eprintln!("Warning: logs command returned no output. Make sure repositories are cloned and contain log files."); } - + for line in stdout_str.lines() { let line = line.trim(); if line.is_empty() { @@ -2199,17 +2229,17 @@ async fn run_build_command(cmd: Command) -> anyhow::Result<()> { } } } - + if total_entries == 0 { eprintln!("Warning: No log entries found. Make sure repositories are cloned and contain log files."); } else if filtered_entries == 0 && !tags_to_use.is_empty() { eprintln!("Warning: Found {} entries but none matched the specified tags. Entries may not have tags yet - consider running 'govbot tag' first, or build without --tags to include all entries.", total_entries); } - + // Deduplicate and sort entries = deduplicate_entries(entries); entries = sort_by_timestamp(entries); - + // Apply limit (default is 100) let original_count = entries.len(); if let Some(lim) = limit_value { @@ -2218,10 +2248,10 @@ async fn run_build_command(cmd: Command) -> anyhow::Result<()> { eprintln!("Limited feed to {} entries (RSS standard). Use --limit none to include all {} entries.", lim, original_count); } } - + // Create output directory fs::create_dir_all(&output_dir_path)?; - + // Generate RSS eprintln!("Generating RSS feed with {} entries...", entries.len()); let rss_xml = rss::json_to_rss( @@ -2232,12 +2262,12 @@ async fn run_build_command(cmd: Command) -> anyhow::Result<()> { base_url.as_deref(), "en-us", ); - + // Write RSS feed let rss_output_path = output_dir_path.join(&output_filename); fs::write(&rss_output_path, rss_xml)?; eprintln!("✓ Generated RSS feed: {}", rss_output_path.display()); - + // Generate HTML eprintln!("Generating HTML index with {} entries...", entries.len()); // Only pass title if it was explicitly set in config (not auto-generated) @@ -2245,48 +2275,49 @@ async fn run_build_command(cmd: Command) -> anyhow::Result<()> { .and_then(|p| p.get("title")) .and_then(|t| t.as_str()) .filter(|s| !s.trim().is_empty()); - let html_content = rss::json_to_html( - entries, - html_title, - feed_link, - base_url.as_deref(), - ); - + let html_content = rss::json_to_html(entries, html_title, feed_link, base_url.as_deref()); + // Write HTML index let html_output_path = output_dir_path.join("index.html"); fs::write(&html_output_path, html_content)?; eprintln!("✓ Generated HTML index: {}", html_output_path.display()); eprintln!(" Tags included: {}", tags_to_use.join(", ")); - + Ok(()) } async fn run_update_command() -> anyhow::Result<()> { let install_script_url = "https://raw.githubusercontent.com/chihacknight/govbot/main/actions/govbot/scripts/install-nightly.sh"; - + eprintln!("🔄 Updating govbot to latest nightly version..."); - eprintln!("Downloading and running install script from: {}", install_script_url); - + eprintln!( + "Downloading and running install script from: {}", + install_script_url + ); + // Execute the install script by piping curl directly to sh // This avoids issues with shebang lines being interpreted as commands let mut cmd = ProcessCommand::new("sh"); cmd.arg("-c"); cmd.arg(&format!("curl -fsSL {} | sh", install_script_url)); - + // Inherit stdin/stdout/stderr so the install script can interact with the user cmd.stdin(std::process::Stdio::inherit()); cmd.stdout(std::process::Stdio::inherit()); cmd.stderr(std::process::Stdio::inherit()); - + let status = cmd.status()?; - + if status.success() { eprintln!("\n✅ Update completed successfully!"); eprintln!("You may need to restart your terminal or run 'source ~/.zshrc' (or your shell profile) to use the updated version."); } else { - return Err(anyhow::anyhow!("Update failed with exit code: {}", status.code().unwrap_or(-1))); + return Err(anyhow::anyhow!( + "Update failed with exit code: {}", + status.code().unwrap_or(-1) + )); } - + Ok(()) } @@ -2295,27 +2326,15 @@ async fn main() -> anyhow::Result<()> { let args = Args::parse(); match args.command { - Some(cmd @ Command::Clone { .. }) => { - run_clone_command(cmd).await - } - Some(cmd @ Command::Delete { .. }) => { - run_delete_command(cmd).await - } - Some(cmd @ Command::Logs { .. }) => { - run_logs_command(cmd).await - } - Some(cmd @ Command::Load { .. }) => { - run_load_command(cmd).await - } - Some(Command::Update) => { - run_update_command().await - } - Some(cmd @ Command::Tag { .. }) => { - run_tag_command(cmd).await - } - Some(cmd @ Command::Build { .. }) => { - run_build_command(cmd).await - } + Some(cmd @ Command::Pull { .. }) => run_clone_command(cmd).await, + Some(cmd @ Command::Delete { .. }) => run_delete_command(cmd).await, + Some(cmd @ Command::Source { .. }) => run_logs_command(cmd).await, + Some(cmd @ Command::Load { .. }) => run_load_command(cmd).await, + Some(Command::Update) => run_update_command().await, + Some(cmd @ Command::Publish { .. }) => run_build_command(cmd).await, + Some(cmd @ Command::Classify { .. }) => run_classify_command(cmd).await, + Some(cmd @ Command::Apply { .. }) => run_apply_command(cmd).await, + Some(cmd @ Command::Run { .. }) => run_run_command(cmd), None => { let cwd = std::env::current_dir()?; let config_path = cwd.join("govbot.yml"); @@ -2330,7 +2349,9 @@ async fn main() -> anyhow::Result<()> { // to start the pipeline (matches the wizard's own message). return Ok(()); } - govbot::pipeline::run_pipeline(&config_path) + // A manifest exists — bare `govbot` runs the default DAG pipeline, + // equivalent to `govbot run`. + govbot::pipeline::run_manifest_pipeline(&config_path, None, false) } } } diff --git a/actions/govbot/src/pipeline.rs b/actions/govbot/src/pipeline.rs index f744cca1..290aafb6 100644 --- a/actions/govbot/src/pipeline.rs +++ b/actions/govbot/src/pipeline.rs @@ -1,137 +1,184 @@ +use crate::config::{Manifest, Transform}; use anyhow::{Context, Result}; use std::path::Path; use std::process::{Command, Stdio}; -/// Run the full govbot pipeline: clone/update → tag → build. +/// Run a manifest pipeline as a DAG: `source | | apply`, then the +/// publisher stages. /// -/// Smart update behavior: -/// - If `.govbot/repos/` exists with repos: just update existing repos (git pull) -/// - If `.govbot/repos/` does not exist: clone repos based on govbot.yml config -pub fn run_pipeline(config_path: &Path) -> Result<()> { - let govbot_bin = std::env::current_exe() - .context("Failed to determine govbot binary path")?; - - let cwd = config_path - .parent() - .unwrap_or_else(|| Path::new(".")); - - let repos_dir = cwd.join(".govbot").join("repos"); - let has_repos = repos_dir.exists() - && std::fs::read_dir(&repos_dir) - .map(|mut d| d.next().is_some()) - .unwrap_or(false); - - // Step 1: Clone or update repos - eprintln!(); - eprintln!("=== Step 1/3: {} repositories ===", if has_repos { "Updating" } else { "Cloning" }); - eprintln!(); - - let clone_status = if has_repos { - // Update existing repos only - Command::new(&govbot_bin) - .arg("clone") - .current_dir(cwd) - .stdin(Stdio::inherit()) - .stdout(Stdio::inherit()) - .stderr(Stdio::inherit()) - .status() +/// The runner is **stage-agnostic**: each `transforms` entry is an opaque +/// subprocess that speaks the stream protocol (newline-JSON on stdio), so the +/// built-in `govbot classify` and an external `fastclass classify -` are peers — +/// swapping one for the other is a manifest `command` edit, not a code change. +/// The schema's typed `reads`/`writes` make branch/merge routing possible later; +/// this release walks the pipeline as a linear chain, which covers the one real +/// `source → classify → apply → publish` shape. +pub fn run_manifest_pipeline( + config_path: &Path, + pipeline_name: Option<&str>, + dry_run: bool, +) -> Result<()> { + let govbot_bin = std::env::current_exe().context("Failed to determine govbot binary path")?; + let cwd = config_path.parent().unwrap_or_else(|| Path::new(".")); + let manifest = Manifest::load(config_path)?; + + // Resolve the ordered stage list: a named pipeline, else the sole/first + // pipeline, else a default of every transform followed by every publisher. + let stages: Vec = if let Some(name) = pipeline_name { + manifest + .pipelines + .get(name) + .cloned() + .ok_or_else(|| anyhow::anyhow!("pipeline '{}' not found in govbot.yml", name))? + } else if let Some((_, first)) = manifest.pipelines.iter().next() { + first.clone() } else { - // First run: clone based on config - let config = crate::publish::load_config(config_path)?; - let repos = crate::publish::get_repos_from_config(&config); - - let mut cmd = Command::new(&govbot_bin); - cmd.arg("clone"); - for repo in &repos { - cmd.arg(repo); - } - cmd.current_dir(cwd) - .stdin(Stdio::inherit()) - .stdout(Stdio::inherit()) - .stderr(Stdio::inherit()) - .status() + let mut s: Vec = manifest.transforms.keys().cloned().collect(); + s.extend(manifest.publish.keys().cloned()); + s }; - match clone_status { - Ok(status) if !status.success() => { - eprintln!("⚠️ Clone/update had errors (continuing anyway)"); - } - Err(e) => { - eprintln!("⚠️ Failed to run clone: {} (continuing anyway)", e); + // Partition into transform stages and publisher stages, preserving order. + let transform_stages: Vec<(String, Transform)> = stages + .iter() + .filter_map(|n| manifest.transforms.get(n).map(|t| (n.clone(), t.clone()))) + .collect(); + let publisher_stages: Vec = stages + .iter() + .filter(|n| manifest.publish.contains_key(*n)) + .cloned() + .collect(); + + // Warn on any stage that names neither a transform nor a publisher. + for stage in &stages { + if !manifest.transforms.contains_key(stage) && !manifest.publish.contains_key(stage) { + eprintln!( + "⚠️ pipeline stage '{}' matches no transform or publisher", + stage + ); } - _ => {} } - // Step 2: Tag bills (govbot logs | govbot tag) + // Step 1: source | transforms… | apply + let chain: Vec<&str> = transform_stages.iter().map(|(n, _)| n.as_str()).collect(); eprintln!(); - eprintln!("=== Step 2/3: Tagging bills ==="); - eprintln!(); - - let tag_result = run_logs_pipe_tag(&govbot_bin, cwd); - match tag_result { - Ok(false) => { - eprintln!("⚠️ Tagging had errors (continuing anyway)"); - } - Err(e) => { - eprintln!("⚠️ Failed to run tagging: {} (continuing anyway)", e); - } - _ => {} - } + eprintln!("=== Transforms: source | {} | apply ===", chain.join(" | ")); + run_transform_chain(&govbot_bin, cwd, &transform_stages)?; - // Step 3: Build RSS feeds - eprintln!(); - eprintln!("=== Step 3/3: Building RSS feeds ==="); + // Step 2: publishers. The per-publisher module split lands in a follow-up; + // for now the publish step is emitted by the existing `govbot build`. eprintln!(); - - let build_status = Command::new(&govbot_bin) - .arg("build") - .current_dir(cwd) - .stdin(Stdio::inherit()) - .stdout(Stdio::inherit()) - .stderr(Stdio::inherit()) - .status() - .context("Failed to run govbot build")?; - - if !build_status.success() { - anyhow::bail!("Build step failed with exit code: {}", build_status.code().unwrap_or(-1)); + eprintln!("=== Publish ==="); + if publisher_stages.is_empty() { + eprintln!("(no publisher stages declared)"); } + run_publish(&govbot_bin, cwd, dry_run)?; eprintln!(); eprintln!("Pipeline complete!"); - Ok(()) } -/// Run `govbot logs | govbot tag` by piping stdout of logs into stdin of tag. -/// Returns Ok(true) if both succeeded, Ok(false) if either failed. -fn run_logs_pipe_tag(govbot_bin: &Path, cwd: &Path) -> Result { - let mut logs_child = Command::new(govbot_bin) - .arg("logs") +/// Spawn `govbot source --select | | govbot apply`, +/// chaining each stage's stdout into the next stage's stdin. +fn run_transform_chain( + govbot_bin: &Path, + cwd: &Path, + transforms: &[(String, Transform)], +) -> Result<()> { + // The projection `source` emits is the first transform's `reads` kind + // (`docs` for a classify pipeline). With no transforms there is nothing to + // classify, so the chain is a no-op. + if transforms.is_empty() { + eprintln!("(no transforms declared — skipping classify stage)"); + return Ok(()); + } + let reads = &transforms[0].1.reads; + + let mut children = Vec::new(); + + let mut source = Command::new(govbot_bin) + .arg("source") + .arg("--select") + .arg(reads) .current_dir(cwd) .stdout(Stdio::piped()) .stderr(Stdio::inherit()) .spawn() - .context("Failed to spawn govbot logs")?; - - let logs_stdout = logs_child - .stdout - .take() - .context("Failed to capture logs stdout")?; + .context("Failed to spawn govbot source")?; + let mut prev_out = source.stdout.take(); + children.push(source); + + for (name, transform) in transforms { + let argv = transform.command.argv(); + if argv.is_empty() { + anyhow::bail!("transform '{}' has an empty command", name); + } + // A `govbot …` transform (the built-in classify) resolves to this same + // executable rather than a `govbot` on PATH; any other program (e.g. + // `fastclass`) is spawned as named — they are otherwise identical. + let program = if argv[0] == "govbot" { + govbot_bin.to_string_lossy().to_string() + } else { + argv[0].clone() + }; + let stdin = prev_out.take().map(Stdio::from).unwrap_or_else(Stdio::null); + let mut child = Command::new(&program) + .args(&argv[1..]) + .current_dir(cwd) + .stdin(stdin) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .spawn() + .with_context(|| format!("Failed to spawn transform '{}' ({})", name, argv[0]))?; + prev_out = child.stdout.take(); + children.push(child); + } - let tag_child = Command::new(govbot_bin) - .arg("tag") + let apply_stdin = prev_out.take().map(Stdio::from).unwrap_or_else(Stdio::null); + let mut apply = Command::new(govbot_bin) + .arg("apply") .current_dir(cwd) - .stdin(logs_stdout) + .stdin(apply_stdin) .stdout(Stdio::inherit()) .stderr(Stdio::inherit()) .spawn() - .context("Failed to spawn govbot tag")?; + .context("Failed to spawn govbot apply")?; + let apply_status = apply.wait().context("Failed to wait for govbot apply")?; - let tag_output = tag_child - .wait_with_output() - .context("Failed to wait for govbot tag")?; + // Reap the upstream stages. + for mut child in children { + let _ = child.wait(); + } - let logs_status = logs_child.wait().context("Failed to wait for govbot logs")?; + if !apply_status.success() { + anyhow::bail!( + "apply stage failed with exit code {}", + apply_status.code().unwrap_or(-1) + ); + } + Ok(()) +} - Ok(logs_status.success() && tag_output.status.success()) +/// Emit the manifest's publishers. For this release publisher stages are emitted +/// by the existing `govbot build` implementation; the per-kind module split (one +/// publisher = one artifact) lands in a follow-up. `--dry-run` is accepted for +/// forward-compatibility and currently just annotates the log line. +fn run_publish(govbot_bin: &Path, cwd: &Path, dry_run: bool) -> Result<()> { + if dry_run { + eprintln!("(dry-run) skipping publisher emission"); + return Ok(()); + } + let status = Command::new(govbot_bin) + .arg("build") + .current_dir(cwd) + .stdin(Stdio::inherit()) + .stdout(Stdio::inherit()) + .stderr(Stdio::inherit()) + .status() + .context("Failed to run publish (build) stage")?; + if !status.success() { + eprintln!("⚠️ publish stage failed (continuing)"); + } + Ok(()) } diff --git a/actions/govbot/src/publish.rs b/actions/govbot/src/publish.rs index 6f9edb31..81adb2e2 100644 --- a/actions/govbot/src/publish.rs +++ b/actions/govbot/src/publish.rs @@ -15,14 +15,17 @@ pub fn load_config(config_path: &Path) -> Result { /// Get repos list from config, handling 'all' special case pub fn get_repos_from_config(config: &Value) -> Vec { - if let Some(repos) = config.get("repos") { - if let Some(arr) = repos.as_array() { - return arr - .iter() - .filter_map(|v| v.as_str().map(|s| s.to_string())) - .collect(); - } else if let Some(s) = repos.as_str() { - return vec![s.to_string()]; + // Prefer the modern `datasets:` key, falling back to the legacy `repos:`. + for key in ["datasets", "repos"] { + if let Some(value) = config.get(key) { + if let Some(arr) = value.as_array() { + return arr + .iter() + .filter_map(|v| v.as_str().map(|s| s.to_string())) + .collect(); + } else if let Some(s) = value.as_str() { + return vec![s.to_string()]; + } } } vec!["all".to_string()] diff --git a/actions/govbot/src/wizard.rs b/actions/govbot/src/wizard.rs index 9368fb37..cea8dabe 100644 --- a/actions/govbot/src/wizard.rs +++ b/actions/govbot/src/wizard.rs @@ -42,12 +42,16 @@ impl WizardSession { display.push('\n'); display.push_str("Available states/jurisdictions:\n"); let all_locales = crate::locale::WorkingLocale::all(); - let locale_strs: Vec = all_locales.iter().map(|l| l.as_str().to_string()).collect(); + let locale_strs: Vec = + all_locales.iter().map(|l| l.as_str().to_string()).collect(); for chunk in locale_strs.chunks(10) { display.push_str(&format!(" {}\n", chunk.join(", "))); } display.push('\n'); - display.push_str(&format!("? Enter state codes separated by spaces: {}\n", choices.repos.join(" "))); + display.push_str(&format!( + "? Enter state codes separated by spaces: {}\n", + choices.repos.join(" ") + )); } display.push('\n'); @@ -77,7 +81,10 @@ impl WizardSession { // Step 3: Publishing display.push_str("Publishing is configured for RSS feeds by default.\n"); display.push_str("Your feeds will be generated in the \"docs\" directory.\n\n"); - display.push_str(&format!("? Base URL for your feeds: {}\n\n", choices.base_url)); + display.push_str(&format!( + "? Base URL for your feeds: {}\n\n", + choices.base_url + )); // Summary display.push_str(" ✓ Created govbot.yml\n"); @@ -85,7 +92,11 @@ impl WizardSession { display.push_str(" ✓ Created .github/workflows/build.yml\n\n"); display.push_str("Setup complete! Run 'govbot' again to start the pipeline.\n"); - let govbot_yml = generate_govbot_yml(&choices.repos, choices.include_example_tag, &choices.base_url); + let govbot_yml = generate_govbot_yml( + &choices.repos, + choices.include_example_tag, + &choices.base_url, + ); let workflow_yml = github_workflow_content().to_string(); WizardSession { @@ -208,10 +219,7 @@ pub fn run_wizard() -> Result<()> { } fn prompt_sources() -> Result> { - let options = vec![ - "All states (47 jurisdictions)", - "Select specific states", - ]; + let options = vec!["All states (47 jurisdictions)", "Select specific states"]; let selection = Select::new() .with_prompt("What data sources do you want to track?") @@ -310,8 +318,8 @@ pub fn generate_govbot_yml(repos: &[String], include_example_tag: bool, base_url yml.push_str("# Schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json\n"); yml.push_str("$schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json\n\n"); - // Repos section - yml.push_str("repos:\n"); + // Datasets section (the DAG's inputs) + yml.push_str("datasets:\n"); for repo in repos { yml.push_str(&format!(" - {}\n", repo)); } @@ -325,9 +333,15 @@ pub fn generate_govbot_yml(repos: &[String], include_example_tag: bool, base_url yml.push_str(" Legislation related to schools, education funding, curriculum standards, and educational policy, including:\n"); yml.push_str(" - K-12 public school funding, budgets, and resource allocation\n"); yml.push_str(" - Curriculum standards, content requirements, and academic programs\n"); - yml.push_str(" - Teacher certification, training, professional development, and compensation\n"); - yml.push_str(" - Higher education policy, tuition, financial aid, and student loans\n"); - yml.push_str(" - Charter schools, school choice, vouchers, and alternative education models\n"); + yml.push_str( + " - Teacher certification, training, professional development, and compensation\n", + ); + yml.push_str( + " - Higher education policy, tuition, financial aid, and student loans\n", + ); + yml.push_str( + " - Charter schools, school choice, vouchers, and alternative education models\n", + ); yml.push_str(" - Special education services, accommodations, and individualized education plans\n"); yml.push_str(" - School safety, security measures, and student discipline policies\n"); yml.push_str(" - Early childhood education, pre-K programs, and childcare\n"); @@ -337,7 +351,9 @@ pub fn generate_govbot_yml(repos: &[String], include_example_tag: bool, base_url yml.push_str(" - Career and technical education, vocational training, and workforce development\n"); yml.push_str(" examples:\n"); yml.push_str(" - \"Increases per-pupil funding for public schools and establishes minimum teacher salary requirements\"\n"); - yml.push_str(" - \"Mandates comprehensive sex education curriculum in all public schools\"\n"); + yml.push_str( + " - \"Mandates comprehensive sex education curriculum in all public schools\"\n", + ); yml.push_str(" - \"Expands eligibility for state financial aid programs to include part-time students\"\n"); } else { yml.push_str(" # Add your tags here. Example:\n"); @@ -350,7 +366,23 @@ pub fn generate_govbot_yml(repos: &[String], include_example_tag: bool, base_url } yml.push('\n'); - // Build section + // Transforms section — the DAG's classify stage. The built-in tagger runs + // as an ordinary transform node; swap `command` to `fastclass classify -` + // (or any stream-protocol program) to change classifiers, no code change. + yml.push_str("transforms:\n"); + yml.push_str(" classify:\n"); + yml.push_str(" command: govbot classify\n"); + yml.push_str(" reads: docs\n"); + yml.push_str(" writes: classification\n"); + yml.push('\n'); + + // Pipelines section — `govbot run` (and bare `govbot`) walks `default`. + yml.push_str("pipelines:\n"); + yml.push_str(" default:\n"); + yml.push_str(" - classify\n"); + yml.push('\n'); + + // Build section (publisher output) yml.push_str("build:\n"); yml.push_str(&format!(" base_url: \"{}\"\n", base_url)); yml.push_str(" output_dir: \"docs\"\n"); diff --git a/actions/govbot/tests/api_snaps.rs b/actions/govbot/tests/api_snaps.rs index 245b302e..536c98d7 100644 --- a/actions/govbot/tests/api_snaps.rs +++ b/actions/govbot/tests/api_snaps.rs @@ -1,10 +1,10 @@ -use govbot::prelude::*; use futures::StreamExt; +use govbot::prelude::*; use insta; /// Snapshot test for the pipeline processor -/// +/// /// This test processes log files and compares the output against stored snapshots. /// To update snapshots after making changes, run: /// cargo insta review @@ -12,7 +12,7 @@ use insta; async fn test_pipeline_processor_snapshot() { // Use the same test data directory as the example let git_dir = "tmp/git/repos"; - + // Build configuration matching the render-snapshots.sh script let config = ConfigBuilder::new(git_dir) .sort_order_str("DESC") @@ -21,7 +21,7 @@ async fn test_pipeline_processor_snapshot() { .join_options_str("bill") .unwrap() .build(); - + // Skip test if git_dir doesn't exist (e.g., in CI without test data) let config = match config { Ok(c) => c, @@ -37,7 +37,7 @@ async fn test_pipeline_processor_snapshot() { // Collect all entries from the stream let mut stream = processor.process(); let mut entries = Vec::new(); - + while let Some(result) = stream.next().await { match result { Ok(entry) => entries.push(entry), @@ -49,8 +49,8 @@ async fn test_pipeline_processor_snapshot() { } // Serialize to JSON for snapshot comparison - let json_output = serde_json::to_string_pretty(&entries) - .expect("Failed to serialize entries to JSON"); + let json_output = + serde_json::to_string_pretty(&entries).expect("Failed to serialize entries to JSON"); // Use insta's assert_snapshot! macro for string comparison // The snapshot will be stored in tests/snapshots/api_snapshot_tests__test_pipeline_processor_snapshot.snap @@ -88,4 +88,3 @@ async fn test_vote_event_processing() { // Test vote event result serialization insta::assert_json_snapshot!("vote_event_results", &results); } - diff --git a/actions/govbot/tests/snapshots/cli_example_snaps__snapshot@govbot_help.snap b/actions/govbot/tests/snapshots/cli_example_snaps__snapshot@govbot_help.snap index 773949e9..f9807842 100644 --- a/actions/govbot/tests/snapshots/cli_example_snaps__snapshot@govbot_help.snap +++ b/actions/govbot/tests/snapshots/cli_example_snaps__snapshot@govbot_help.snap @@ -1,5 +1,6 @@ --- source: tests/cli_example_snaps.rs +assertion_line: 223 expression: "&formatted_stdout" --- Command: @@ -11,14 +12,16 @@ Process pipeline log files with type-safe reactive streams Usage: govbot [COMMAND] Commands: - clone Clone or pull data pipeline repositories (default: updates existing repos) Clones if repository doesn't exist, pulls if it does Use "govbot clone all" to clone all repos, or "govbot clone " for specific repos - logs Process and display pipeline log files - delete Delete data pipeline repositories Deletes local repository directories for specified locales - load Load bill metadata into a DuckDB database file Loads all metadata.json files from cloned repos into a DuckDB database for analysis. The database file is saved in the base govbot directory (e.g., ./.govbot/govbot.duckdb) - update Update govbot to the latest nightly version Downloads and installs the latest nightly build from GitHub releases - build Build RSS feed and HTML index from govbot.yml configuration Generates a combined RSS feed and HTML index from logs filtered by tags in govbot.yml - tag Tag bills using semantic or built-in similarity based on govbot.yml in the current directory. Reads JSON lines from stdin (from `govbot logs`), processes entries with bill identifiers, and writes per-tag files under the directory containing govbot.yml. By default, acts as a filter: only outputs lines that match tags. If a tag name is provided, only processes and outputs lines matching that specific tag - help Print this message or the help of the given subcommand(s) + pull Pull data pipeline datasets (default: updates existing datasets) Clones if a dataset doesn't exist, pulls if it does. Use "govbot pull all" for all datasets, or "govbot pull " for specific ones [aliases: clone] + source Emit the activity stream (the DAG's `source` stage) [aliases: logs] + delete Delete data pipeline repositories Deletes local repository directories for specified locales + load Load bill metadata into a DuckDB database file Loads all metadata.json files from cloned repos into a DuckDB database for analysis. The database file is saved in the base govbot directory (e.g., ./.govbot/govbot.duckdb) + update Update govbot to the latest nightly version Downloads and installs the latest nightly build from GitHub releases + publish Publish RSS feed and HTML index from govbot.yml (the DAG's publish stage) Generates a combined RSS feed and HTML index from the stream filtered by tags [aliases: build] + classify Classify records from stdin with the built-in tagger (a DAG transform) + apply Apply classification records from stdin, persisting per-tag .tag.json files + run Run a manifest pipeline: source | transforms | apply, then publishers + help Print this message or the help of the given subcommand(s) Options: -h, --help Print help diff --git a/actions/govbot/tests/snapshots/wizard_tests__wizard_all_no_tag.snap b/actions/govbot/tests/snapshots/wizard_tests__wizard_all_no_tag.snap index a3ae97fb..3fb474eb 100644 --- a/actions/govbot/tests/snapshots/wizard_tests__wizard_all_no_tag.snap +++ b/actions/govbot/tests/snapshots/wizard_tests__wizard_all_no_tag.snap @@ -1,12 +1,13 @@ --- source: tests/wizard_tests.rs +assertion_line: 113 expression: "&yml" --- # Govbot Configuration # Schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json $schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json -repos: +datasets: - all tags: @@ -18,6 +19,16 @@ tags: # - "Example bill description" {} +transforms: + classify: + command: govbot classify + reads: docs + writes: classification + +pipelines: + default: + - classify + build: base_url: "https://example.com" output_dir: "docs" diff --git a/actions/govbot/tests/snapshots/wizard_tests__wizard_all_with_tag.snap b/actions/govbot/tests/snapshots/wizard_tests__wizard_all_with_tag.snap index df8d77bf..8262e29a 100644 --- a/actions/govbot/tests/snapshots/wizard_tests__wizard_all_with_tag.snap +++ b/actions/govbot/tests/snapshots/wizard_tests__wizard_all_with_tag.snap @@ -1,12 +1,13 @@ --- source: tests/wizard_tests.rs +assertion_line: 89 expression: "&yml" --- # Govbot Configuration # Schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json $schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json -repos: +datasets: - all tags: @@ -30,6 +31,16 @@ tags: - "Mandates comprehensive sex education curriculum in all public schools" - "Expands eligibility for state financial aid programs to include part-time students" +transforms: + classify: + command: govbot classify + reads: docs + writes: classification + +pipelines: + default: + - classify + build: base_url: "https://myuser.github.io/my-govbot" output_dir: "docs" diff --git a/actions/govbot/tests/snapshots/wizard_tests__wizard_session_all_own_tags.snap b/actions/govbot/tests/snapshots/wizard_tests__wizard_session_all_own_tags.snap index 127602d2..000c48a1 100644 --- a/actions/govbot/tests/snapshots/wizard_tests__wizard_session_all_own_tags.snap +++ b/actions/govbot/tests/snapshots/wizard_tests__wizard_session_all_own_tags.snap @@ -1,5 +1,6 @@ --- source: tests/wizard_tests.rs +assertion_line: 33 expression: "&session.to_snapshot()" --- === Wizard Session === @@ -65,7 +66,7 @@ Setup complete! Run 'govbot' again to start the pipeline. # Schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json $schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json -repos: +datasets: - all tags: @@ -77,6 +78,16 @@ tags: # - "Example bill description" {} +transforms: + classify: + command: govbot classify + reads: docs + writes: classification + +pipelines: + default: + - classify + build: base_url: "https://example.com" output_dir: "docs" diff --git a/actions/govbot/tests/snapshots/wizard_tests__wizard_session_all_with_tag.snap b/actions/govbot/tests/snapshots/wizard_tests__wizard_session_all_with_tag.snap index b0d13d03..d42b5252 100644 --- a/actions/govbot/tests/snapshots/wizard_tests__wizard_session_all_with_tag.snap +++ b/actions/govbot/tests/snapshots/wizard_tests__wizard_session_all_with_tag.snap @@ -1,5 +1,6 @@ --- source: tests/wizard_tests.rs +assertion_line: 19 expression: "&session.to_snapshot()" --- === Wizard Session === @@ -42,7 +43,7 @@ Setup complete! Run 'govbot' again to start the pipeline. # Schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json $schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json -repos: +datasets: - all tags: @@ -66,6 +67,16 @@ tags: - "Mandates comprehensive sex education curriculum in all public schools" - "Expands eligibility for state financial aid programs to include part-time students" +transforms: + classify: + command: govbot classify + reads: docs + writes: classification + +pipelines: + default: + - classify + build: base_url: "https://myuser.github.io/my-govbot" output_dir: "docs" diff --git a/actions/govbot/tests/snapshots/wizard_tests__wizard_session_single_state.snap b/actions/govbot/tests/snapshots/wizard_tests__wizard_session_single_state.snap index e5413262..acb04298 100644 --- a/actions/govbot/tests/snapshots/wizard_tests__wizard_session_single_state.snap +++ b/actions/govbot/tests/snapshots/wizard_tests__wizard_session_single_state.snap @@ -1,5 +1,6 @@ --- source: tests/wizard_tests.rs +assertion_line: 75 expression: "&session.to_snapshot()" --- === Wizard Session === @@ -52,7 +53,7 @@ Setup complete! Run 'govbot' again to start the pipeline. # Schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json $schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json -repos: +datasets: - wy tags: @@ -76,6 +77,16 @@ tags: - "Mandates comprehensive sex education curriculum in all public schools" - "Expands eligibility for state financial aid programs to include part-time students" +transforms: + classify: + command: govbot classify + reads: docs + writes: classification + +pipelines: + default: + - classify + build: base_url: "https://sartaj.me/govbot" output_dir: "docs" diff --git a/actions/govbot/tests/snapshots/wizard_tests__wizard_session_specific_own_tags.snap b/actions/govbot/tests/snapshots/wizard_tests__wizard_session_specific_own_tags.snap index 727838b5..230851cf 100644 --- a/actions/govbot/tests/snapshots/wizard_tests__wizard_session_specific_own_tags.snap +++ b/actions/govbot/tests/snapshots/wizard_tests__wizard_session_specific_own_tags.snap @@ -1,5 +1,6 @@ --- source: tests/wizard_tests.rs +assertion_line: 61 expression: "&session.to_snapshot()" --- === Wizard Session === @@ -75,7 +76,7 @@ Setup complete! Run 'govbot' again to start the pipeline. # Schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json $schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json -repos: +datasets: - il - ca - ny @@ -89,6 +90,16 @@ tags: # - "Example bill description" {} +transforms: + classify: + command: govbot classify + reads: docs + writes: classification + +pipelines: + default: + - classify + build: base_url: "https://example.com" output_dir: "docs" diff --git a/actions/govbot/tests/snapshots/wizard_tests__wizard_session_specific_with_tag.snap b/actions/govbot/tests/snapshots/wizard_tests__wizard_session_specific_with_tag.snap index 528b4933..e50949bb 100644 --- a/actions/govbot/tests/snapshots/wizard_tests__wizard_session_specific_with_tag.snap +++ b/actions/govbot/tests/snapshots/wizard_tests__wizard_session_specific_with_tag.snap @@ -1,5 +1,6 @@ --- source: tests/wizard_tests.rs +assertion_line: 47 expression: "&session.to_snapshot()" --- === Wizard Session === @@ -52,7 +53,7 @@ Setup complete! Run 'govbot' again to start the pipeline. # Schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json $schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json -repos: +datasets: - il - ca - ny @@ -78,6 +79,16 @@ tags: - "Mandates comprehensive sex education curriculum in all public schools" - "Expands eligibility for state financial aid programs to include part-time students" +transforms: + classify: + command: govbot classify + reads: docs + writes: classification + +pipelines: + default: + - classify + build: base_url: "https://activist.github.io/legislation" output_dir: "docs" diff --git a/actions/govbot/tests/snapshots/wizard_tests__wizard_single_with_tag.snap b/actions/govbot/tests/snapshots/wizard_tests__wizard_single_with_tag.snap index 3e513413..bb34aee9 100644 --- a/actions/govbot/tests/snapshots/wizard_tests__wizard_single_with_tag.snap +++ b/actions/govbot/tests/snapshots/wizard_tests__wizard_single_with_tag.snap @@ -1,12 +1,13 @@ --- source: tests/wizard_tests.rs +assertion_line: 123 expression: "&yml" --- # Govbot Configuration # Schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json $schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json -repos: +datasets: - wy tags: @@ -30,6 +31,16 @@ tags: - "Mandates comprehensive sex education curriculum in all public schools" - "Expands eligibility for state financial aid programs to include part-time students" +transforms: + classify: + command: govbot classify + reads: docs + writes: classification + +pipelines: + default: + - classify + build: base_url: "https://sartaj.me/govbot" output_dir: "docs" diff --git a/actions/govbot/tests/snapshots/wizard_tests__wizard_specific_no_tag.snap b/actions/govbot/tests/snapshots/wizard_tests__wizard_specific_no_tag.snap index f3ab59ea..8fc5d972 100644 --- a/actions/govbot/tests/snapshots/wizard_tests__wizard_specific_no_tag.snap +++ b/actions/govbot/tests/snapshots/wizard_tests__wizard_specific_no_tag.snap @@ -1,12 +1,13 @@ --- source: tests/wizard_tests.rs +assertion_line: 103 expression: "&yml" --- # Govbot Configuration # Schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json $schema: https://raw.githubusercontent.com/chihacknight/govbot/main/schemas/govbot.schema.json -repos: +datasets: - il - ca - ny @@ -20,6 +21,16 @@ tags: # - "Example bill description" {} +transforms: + classify: + command: govbot classify + reads: docs + writes: classification + +pipelines: + default: + - classify + build: base_url: "https://example.com" output_dir: "docs" diff --git a/actions/govbot/tests/wizard_tests.rs b/actions/govbot/tests/wizard_tests.rs index e8f30725..f3ba2567 100644 --- a/actions/govbot/tests/wizard_tests.rs +++ b/actions/govbot/tests/wizard_tests.rs @@ -1,5 +1,5 @@ +use govbot::publish::{get_repos_from_config, load_config}; use govbot::wizard::{generate_govbot_yml, WizardChoices, WizardSession}; -use govbot::publish::{load_config, get_repos_from_config}; // ============================================================ // Full wizard session snapshots — shows the entire user experience @@ -82,7 +82,11 @@ fn wizard_session_single_state() { #[test] fn test_generate_govbot_yml_all_repos_with_example_tag() { - let yml = generate_govbot_yml(&["all".to_string()], true, "https://myuser.github.io/my-govbot"); + let yml = generate_govbot_yml( + &["all".to_string()], + true, + "https://myuser.github.io/my-govbot", + ); let mut settings = insta::Settings::clone_current(); settings.set_snapshot_path("snapshots"); settings.bind(|| { @@ -131,7 +135,11 @@ fn test_generate_govbot_yml_single_repo_with_tag() { #[test] fn test_generated_yml_is_valid_yaml_with_tag() { - let yml = generate_govbot_yml(&["all".to_string()], true, "https://myuser.github.io/my-govbot"); + let yml = generate_govbot_yml( + &["all".to_string()], + true, + "https://myuser.github.io/my-govbot", + ); let dir = tempfile::tempdir().unwrap(); let config_path = dir.path().join("govbot.yml"); std::fs::write(&config_path, &yml).unwrap(); @@ -145,17 +153,35 @@ fn test_generated_yml_is_valid_yaml_with_tag() { // Verify tags exist and have expected structure let tags = config.get("tags").expect("should have tags key"); let tags_obj = tags.as_object().expect("tags should be an object"); - assert!(tags_obj.contains_key("education"), "should contain education tag"); + assert!( + tags_obj.contains_key("education"), + "should contain education tag" + ); let education = tags_obj.get("education").unwrap().as_object().unwrap(); - assert!(education.contains_key("description"), "education tag should have description"); - assert!(education.contains_key("examples"), "education tag should have examples"); + assert!( + education.contains_key("description"), + "education tag should have description" + ); + assert!( + education.contains_key("examples"), + "education tag should have examples" + ); // Verify build config let build = config.get("build").expect("should have build key"); let build_obj = build.as_object().expect("build should be an object"); - assert_eq!(build_obj.get("base_url").unwrap().as_str().unwrap(), "https://myuser.github.io/my-govbot"); - assert_eq!(build_obj.get("output_dir").unwrap().as_str().unwrap(), "docs"); - assert_eq!(build_obj.get("output_file").unwrap().as_str().unwrap(), "feed.xml"); + assert_eq!( + build_obj.get("base_url").unwrap().as_str().unwrap(), + "https://myuser.github.io/my-govbot" + ); + assert_eq!( + build_obj.get("output_dir").unwrap().as_str().unwrap(), + "docs" + ); + assert_eq!( + build_obj.get("output_file").unwrap().as_str().unwrap(), + "feed.xml" + ); } #[test] @@ -178,12 +204,18 @@ fn test_generated_yml_is_valid_yaml_without_tag() { // Verify tags is empty object let tags = config.get("tags").expect("should have tags key"); let tags_obj = tags.as_object().expect("tags should be an object"); - assert!(tags_obj.is_empty(), "tags should be empty when no example tag"); + assert!( + tags_obj.is_empty(), + "tags should be empty when no example tag" + ); // Verify build config let build = config.get("build").expect("should have build key"); let build_obj = build.as_object().expect("build should be an object"); - assert_eq!(build_obj.get("base_url").unwrap().as_str().unwrap(), "https://example.com"); + assert_eq!( + build_obj.get("base_url").unwrap().as_str().unwrap(), + "https://example.com" + ); } #[test] @@ -196,7 +228,9 @@ fn test_write_files_creates_govbot_yml() { let session = WizardSession::from_choices(&choices); let dir = tempfile::tempdir().unwrap(); - session.write_files(dir.path()).expect("write_files should succeed"); + session + .write_files(dir.path()) + .expect("write_files should succeed"); // Verify govbot.yml was created and is parseable let config_path = dir.path().join("govbot.yml"); @@ -209,7 +243,10 @@ fn test_write_files_creates_govbot_yml() { let gitignore_path = dir.path().join(".gitignore"); assert!(gitignore_path.exists(), ".gitignore should exist"); let gitignore = std::fs::read_to_string(&gitignore_path).unwrap(); - assert!(gitignore.contains(".govbot"), ".gitignore should contain .govbot"); + assert!( + gitignore.contains(".govbot"), + ".gitignore should contain .govbot" + ); // Verify workflow was created let workflow_path = dir.path().join(".github/workflows/build.yml"); diff --git a/schemas/govbot.schema.json b/schemas/govbot.schema.json index 555a2318..78ca02b8 100644 --- a/schemas/govbot.schema.json +++ b/schemas/govbot.schema.json @@ -1,61 +1,142 @@ { "$schema": "http://json-schema.org/draft-07/schema#", - "title": "Govbot Configuration Schema", - "description": "Schema for validating govbot.yml configuration files", + "title": "Govbot Manifest Schema", + "description": "Schema for validating govbot.yml manifest files. govbot.yml declares the datasets a project consumes, the transforms it runs over them, the publishers that emit artifacts, and named pipelines that wire those stages into a DAG. A transform node is a uniform { command, reads, writes }: the built-in tagger fills the classify role as an ordinary transform whose command is 'govbot classify', and swapping in an external classifier (e.g. 'fastclass classify -') is only a command change. The legacy 'repos' and 'tags' keys are still accepted for backwards compatibility.", "type": "object", "properties": { + "$schema": { + "type": "string" + }, + "datasets": { + "description": "Datasets the project consumes (git repos / registry names). Use 'all' to include every available dataset. Additive superset of the legacy 'repos'.", + "type": "array", + "items": { + "type": "string" + } + }, "repos": { - "description": "List of repositories to clone and process. Use 'all' to include all available repositories.", + "description": "Legacy dataset list, still honored. Prefer 'datasets'.", "type": "array", "items": { "type": "string" - }, - "default": ["all"] + } }, "tags": { - "description": "Tag definitions for categorizing legislation. Each tag should have a description and optional examples.", + "description": "Tag taxonomy consumed by the built-in 'govbot classify' transform. Each tag has a description and optional examples/keywords. An external classifier transform (e.g. fastclass) carries its own taxonomy instead.", "type": "object", "additionalProperties": { "$ref": "#/definitions/tag" } }, + "transforms": { + "description": "Named external-process transform nodes. A transform is a separate program that speaks the govbot stream protocol (newline-delimited JSON on stdio, stable 'id', typed 'kind'). govbot streams records of the transform's 'reads' kind into it and routes records of its 'writes' kind back by 'id'. 'govbot classify' (the built-in tagger) and 'fastclass classify -' are interchangeable transform nodes.", + "type": "object", + "additionalProperties": { + "$ref": "#/definitions/transform" + } + }, "publish": { - "description": "RSS feed publishing configuration", + "description": "Named publisher nodes. Each publisher consumes the result stream and emits exactly one artifact. Declare one publisher per artifact kind (e.g. both an 'rss' and an 'html' publisher to emit both a feed and an index).", + "type": "object", + "additionalProperties": { + "$ref": "#/definitions/publisher" + } + }, + "pipelines": { + "description": "Named 'govbot run' targets, npm-script style. Each pipeline is an ordered list of stage references -- names of entries in 'transforms' and 'publish' -- executed in sequence. 'govbot run ' runs the named pipeline.", + "type": "object", + "additionalProperties": { + "$ref": "#/definitions/pipeline" + } + } + }, + "definitions": { + "tag": { "type": "object", "properties": { - "base_url": { - "description": "Base URL for RSS feed links (required for GitHub Pages). Should match your GitHub Pages URL.", - "type": "string", - "format": "uri", - "default": "https://example.com" + "description": { + "description": "Detailed description of what legislation this tag covers. Use YAML multiline strings (|) for formatting.", + "type": "string" }, - "output_dir": { - "description": "Directory where RSS feeds are generated", + "examples": { + "description": "Example bill descriptions that would match this tag", + "type": "array", + "items": { + "type": "string" + } + } + }, + "required": ["description"] + }, + "transform": { + "description": "A single external-process transform stage (a DAG node). Uniform shape -- no classify-specific fields; a classifier bundle path, if any, is part of 'command'.", + "type": "object", + "properties": { + "command": { + "description": "The stage command: a shell string ('govbot classify') or an explicit argv array (['govbot', 'classify']).", + "oneOf": [ + { + "type": "string" + }, + { + "type": "array", + "items": { + "type": "string" + } + } + ] + }, + "reads": { + "description": "The stream record kind this transform consumes. 'docs' is the document-projection kind defined by the stream protocol.", "type": "string", - "default": "feeds" + "examples": ["docs", "classification", "summary"] }, - "output_file": { - "description": "Output filename for the RSS feed", + "writes": { + "description": "The stream record kind this transform produces. govbot routes records of this kind back into the stream by their 'id'. The classify transform writes 'classification'.", + "type": "string", + "examples": ["classification", "summary"] + } + }, + "required": ["command", "reads", "writes"] + }, + "publisher": { + "description": "A single publisher stage. Each publisher kind emits exactly one artifact: 'rss' writes the RSS feed (default feed.xml), 'html' writes the HTML index (default index.html).", + "type": "object", + "properties": { + "type": { + "description": "The publisher kind. 'rss' writes the RSS feed only; 'html' writes the HTML index only.", "type": "string", - "default": "feed.xml" + "enum": ["rss", "html"] }, - "tags": { - "description": "Specific tags to include in the combined RSS feed. If not specified, all tags are included.", + "select": { + "description": "Tag names to include. Only records carrying at least one of these tags are published; if omitted, all tagged records are published.", "type": "array", "items": { "type": "string" } }, + "base_url": { + "description": "Base URL for generated links (e.g. the GitHub Pages URL).", + "type": "string" + }, + "output_dir": { + "description": "Directory where the publisher writes its artifact.", + "type": "string" + }, + "output_file": { + "description": "Output filename for the publisher's single artifact. Defaults by 'type': 'rss' -> 'feed.xml', 'html' -> 'index.html'.", + "type": "string" + }, "title": { - "description": "Custom feed title. If not specified, defaults to combined tag names.", + "description": "Custom feed/page title.", "type": "string" }, "description": { - "description": "Custom feed description. If not specified, defaults to combined tag descriptions.", + "description": "Custom feed/page description.", "type": "string" }, "limit": { - "description": "Limit number of entries per RSS feed. Use 'none' for no limit, or a number. Default is 15 (RSS standard).", + "description": "Limit number of entries. Use 'none' for no limit, or a number.", "oneOf": [ { "type": "string", @@ -68,28 +149,14 @@ ] } }, - "required": ["base_url", "output_dir", "output_file"] - } - }, - "required": ["repos", "tags"], - "definitions": { - "tag": { - "type": "object", - "properties": { - "description": { - "description": "Detailed description of what legislation this tag covers. Use YAML multiline strings (|) for formatting.", - "type": "string" - }, - "examples": { - "description": "Example bill descriptions that would match this tag", - "type": "array", - "items": { - "type": "string" - } - } - }, - "required": ["description"] + "required": ["type"] + }, + "pipeline": { + "description": "An ordered list of stage references (names of 'transforms' and 'publish' entries) executed in sequence.", + "type": "array", + "items": { + "type": "string" + } } } } -