Make Govbot a DAG

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

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:

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 <pipeline>` 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<String>,
+
+    /// Datasets the project consumes. Additive superset of the legacy `repos:`.
+    #[serde(default)]
+    pub datasets: Vec<String>,
+
+    /// Legacy dataset list; still honored so old manifests keep working.
+    #[serde(default)]
+    pub repos: Vec<String>,
+
+    /// Named external-process transform nodes. Uniform shape — no privileged
+    /// classify node; `govbot classify` and `fastclass classify -` are peers.
+    #[serde(default)]
+    pub transforms: BTreeMap<String, Transform>,
+
+    /// 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<String, Publisher>,
+
+    /// Named `govbot run` targets: each an ordered list of stage names that
+    /// reference entries in `transforms` and `publish`.
+    #[serde(default)]
+    pub pipelines: BTreeMap<String, Vec<String>>,
+}
+
+impl Manifest {
+    /// Load and parse a `govbot.yml` manifest from disk.
+    pub fn load(path: impl AsRef<Path>) -> Result<Self> {
+        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<String>),
+}
+
+impl CommandSpec {
+    /// The command split into program + arguments.
+    pub fn argv(&self) -> Vec<String> {
+        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<Vec<String>>,
+
+    /// Base URL for generated links (e.g. the GitHub Pages URL).
+    #[serde(default)]
+    pub base_url: Option<String>,
+
+    /// Directory the publisher writes its artifact to.
+    #[serde(default)]
+    pub output_dir: Option<String>,
+
+    /// Output filename; defaults by kind (`rss` -> feed.xml, `html` -> index.html).
+    #[serde(default)]
+    pub output_file: Option<String>,
+
+    #[serde(default)]
+    pub title: Option<String>,
+
+    #[serde(default)]
+    pub description: Option<String>,
+
+    /// Max entries; a number, or the string `none` for all.
+    #[serde(default)]
+    pub limit: Option<serde_yaml::Value>,
+}
 
 /// 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()]);
+    }
+}

actions/govbot/src/embeddings.rs

@@ -404,15 +404,25 @@ impl TagMatcher {
         &self,
         value: &serde_json::Value,
     ) -> anyhow::Result<Vec<(String, ScoreBreakdown)>> {
-        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<Vec<(String, ScoreBreakdown)>> {
         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();