Merge pull request #36 from Digital-Threads/feat/memory-consolidation-p32

feat(memory): consolidation via direct Haiku API — Pillar C complete (0.19.0)

Author: Shahinyanm

CHANGELOG.md

@@ -7,6 +7,32 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.19.0] - 2026-06-13
+
+### Added
+- **Consolidation — Pillar C complete.** `task-journal consolidate` distils a
+  project's recurring decisions and constraints into a handful of durable
+  **semantic** / **procedural** facts ("refunds always go through the ledger",
+  "PR into main, squash-merge"), stored as events in a per-project
+  *"Project conventions (consolidated)"* task with provenance
+  (`derived_from`), and surfaced in `ask`/`recall`.
+  - **Manual and opt-in.** It makes exactly **one direct Anthropic Haiku API
+    call per run, only when you run it** — never wired to a hook, so it can
+    never spend automatically. Roughly 1¢ per run.
+  - The **direct API** (not `claude -p`) is used on purpose: post-2026-06-15
+    both bill as extra usage, but `claude -p` also boots the whole environment
+    (~tens of k tokens) per call; the direct API sends only the ~7k-token
+    prompt.
+  - **No `ANTHROPIC_API_KEY` → it skips cleanly** with a message and writes
+    nothing. There is no heuristic fallback (it would manufacture low-trust
+    facts). Re-running de-duplicates.
+  - `--max-facts N` caps output; `TJ_CONSOLIDATE_MODEL` overrides the model.
+
+### Internal
+- `tj-core::consolidate` (prompt, parse, direct-API call, mockito-tested) +
+  `db::high_signal_events` / `find_task_by_title` / `task_event_texts`. CLI
+  `consolidate`.
+
 ## [0.18.0] - 2026-06-12
 
 ### Added

Cargo.lock

@@ -7,7 +7,7 @@ members = [
 ]
 
 [workspace.package]
-version = "0.18.0"
+version = "0.19.0"
 edition = "2021"
 rust-version = "1.88"
 license = "MIT"

Cargo.toml

@@ -23,7 +23,7 @@ default = ["embed"]
 embed = ["tj-core/embed"]
 
 [dependencies]
-tj-core = { package = "task-journal-core", version = "0.18.0", path = "../tj-core", default-features = false }
+tj-core = { package = "task-journal-core", version = "0.19.0", path = "../tj-core", default-features = false }
 anyhow = { workspace = true }
 clap = { workspace = true }
 tracing = { workspace = true }
@@ -45,3 +45,4 @@ assert_fs = { workspace = true }
 predicates = { workspace = true }
 assert_cmd = ">=2, <2.2.1"
 rusqlite = { workspace = true }
+mockito = { workspace = true }

crates/tj-cli/Cargo.toml

@@ -652,6 +652,16 @@ enum Commands {
     },
     /// List your stored user preferences.
     Preferences,
+    /// Distil this project's recurring decisions and constraints into durable
+    /// semantic/procedural facts (Pillar C). MANUAL and opt-in — it makes ONE
+    /// direct Haiku API call per run (needs ANTHROPIC_API_KEY; ~1c/run) and is
+    /// never wired to a hook, so it can't spend automatically. Facts are stored
+    /// as events in a per-project "conventions" task and surface in ask/recall.
+    Consolidate {
+        /// Maximum number of facts to produce.
+        #[arg(long, default_value_t = 8)]
+        max_facts: usize,
+    },
     /// Render and print the resume pack for a task.
     Pack {
         /// Task id (e.g. tj-7f3a).
@@ -1284,6 +1294,9 @@ fn main() -> Result<()> {
                 }
             }
         }
+        Commands::Consolidate { max_facts } => {
+            run_consolidate(max_facts)?;
+        }
         Commands::Event {
             task_id,
             r#type,
@@ -3904,6 +3917,112 @@ fn emit_session_context(ctx: &str) {
     println!("{env}");
 }
 
+const CONSOLIDATE_TASK_TITLE: &str = "Project conventions (consolidated)";
+
+/// Manual consolidation: read this project's recurring decisions/constraints,
+/// distil them into durable facts via one direct Haiku API call, and store the
+/// facts as events in a per-project conventions task. Skips cleanly (no spend)
+/// when ANTHROPIC_API_KEY is absent.
+fn run_consolidate(max_facts: usize) -> anyhow::Result<()> {
+    let cwd = std::env::current_dir()?;
+    let project_hash = tj_core::project_hash::from_path(&cwd)?;
+    let events_path = tj_core::paths::events_dir()?.join(format!("{project_hash}.jsonl"));
+    let state_path = tj_core::paths::state_dir()?.join(format!("{project_hash}.sqlite"));
+    if !events_path.exists() {
+        anyhow::bail!("no events file at {events_path:?}");
+    }
+    let conn = tj_core::db::open(&state_path)?;
+    tj_core::db::ingest_new_events(&conn, &events_path, &project_hash)?;
+
+    let sources = tj_core::db::high_signal_events(&conn, 200)?;
+    if sources.is_empty() {
+        println!("nothing to consolidate — no decisions/constraints/rejections recorded yet");
+        return Ok(());
+    }
+    let texts: Vec<String> = sources.iter().map(|(_, t)| t.clone()).collect();
+    let source_ids: Vec<String> = sources.iter().map(|(id, _)| id.clone()).collect();
+
+    let consolidator = match tj_core::consolidate::Consolidator::from_env(max_facts) {
+        Ok(c) => c,
+        Err(e) => {
+            println!("skipped: {e}. Set ANTHROPIC_API_KEY to enable consolidation (~1c/run).");
+            return Ok(());
+        }
+    };
+    eprintln!(
+        "consolidating {} high-signal event(s) via {} …",
+        texts.len(),
+        consolidator.model
+    );
+    let facts = consolidator.consolidate(&texts)?;
+    if facts.is_empty() {
+        println!("no durable facts found");
+        return Ok(());
+    }
+
+    // Reuse the per-project conventions task, or create it.
+    let task_id = match tj_core::db::find_task_by_title(&conn, CONSOLIDATE_TASK_TITLE)? {
+        Some(id) => id,
+        None => {
+            let id = tj_core::new_task_id();
+            let mut ev = tj_core::event::Event::new(
+                id.clone(),
+                tj_core::event::EventType::Open,
+                tj_core::event::Author::User,
+                tj_core::event::Source::Cli,
+                CONSOLIDATE_TASK_TITLE.to_string(),
+            );
+            ev.meta = serde_json::json!({ "title": CONSOLIDATE_TASK_TITLE });
+            let mut w = tj_core::storage::JsonlWriter::open(&events_path)?;
+            w.append(&ev)?;
+            w.flush_durable()?;
+            tj_core::db::ingest_new_events(&conn, &events_path, &project_hash)?;
+            id
+        }
+    };
+
+    // De-dup against facts already stored in the conventions task.
+    let existing: std::collections::HashSet<String> =
+        tj_core::db::task_event_texts(&conn, &task_id)?
+            .into_iter()
+            .collect();
+
+    let mut writer = tj_core::storage::JsonlWriter::open(&events_path)?;
+    let mut written = 0usize;
+    for f in &facts {
+        if existing.contains(&f.text) {
+            continue;
+        }
+        let mut ev = tj_core::event::Event::new(
+            task_id.clone(),
+            tj_core::event::EventType::Finding,
+            tj_core::event::Author::Agent,
+            tj_core::event::Source::Cli,
+            f.text.clone(),
+        );
+        ev.meta = serde_json::json!({
+            "memory_tier": f.tier,
+            "consolidated": true,
+            "derived_from": source_ids,
+        });
+        writer.append(&ev)?;
+        written += 1;
+    }
+    writer.flush_durable()?;
+
+    // Index the new facts and push them to the global recall index.
+    tj_core::db::ingest_new_events(&conn, &events_path, &project_hash)?;
+    let embedder = tj_core::embed::default_embedder();
+    let now = chrono::Utc::now().to_rfc3339();
+    tj_core::db::embed_pending(&conn, &project_hash, embedder.as_ref(), &now, 512)?;
+    sync_global_memory(&conn, &project_hash);
+
+    println!(
+        "consolidated {written} new fact(s) into task {task_id} (\"{CONSOLIDATE_TASK_TITLE}\")"
+    );
+    Ok(())
+}
+
 fn auto_open_task_from_prompt(
     events_path: &std::path::Path,
     project_hash: &str,

crates/tj-cli/src/main.rs

@@ -5209,3 +5209,154 @@ fn stats_reports_memory_preferences_count() {
         .success()
         .stdout(contains("preferences: 1"));
 }
+
+#[test]
+fn consolidate_writes_facts_to_conventions_task_and_dedups() {
+    // Pillar C: `consolidate` distils decisions into durable facts via one
+    // (mocked) Haiku call and stores them in a per-project conventions task.
+    // Re-running de-dups. TJ_CONSOLIDATE_BASE_URL points at the mock; TJ_EMBED
+    // forces the deterministic embedder.
+    let mut server = mockito::Server::new();
+    let mock = server
+        .mock("POST", "/v1/messages")
+        .with_status(200)
+        .with_header("content-type", "application/json")
+        .with_body(
+            serde_json::json!({
+                "id": "m", "type": "message", "role": "assistant",
+                "content": [{"type": "text",
+                    "text": "[semantic] Refunds always route through the idempotent ledger\n[procedural] PR into main, squash-merge"}]
+            })
+            .to_string(),
+        )
+        .expect_at_least(1)
+        .create();
+
+    let xdg = assert_fs::TempDir::new().unwrap();
+    let proj = assert_fs::TempDir::new().unwrap();
+
+    let tid = String::from_utf8(
+        Command::cargo_bin("task-journal")
+            .unwrap()
+            .current_dir(proj.path())
+            .env("XDG_DATA_HOME", xdg.path())
+            .args(["create", "Payments"])
+            .assert()
+            .success()
+            .get_output()
+            .stdout
+            .clone(),
+    )
+    .unwrap()
+    .trim()
+    .to_string();
+    Command::cargo_bin("task-journal")
+        .unwrap()
+        .current_dir(proj.path())
+        .env("XDG_DATA_HOME", xdg.path())
+        .args([
+            "event",
+            &tid,
+            "--type",
+            "decision",
+            "--text",
+            "chose the idempotent ledger for refunds",
+        ])
+        .assert()
+        .success();
+
+    let run = || {
+        Command::cargo_bin("task-journal")
+            .unwrap()
+            .current_dir(proj.path())
+            .env("XDG_DATA_HOME", xdg.path())
+            .env("ANTHROPIC_API_KEY", "test-key")
+            .env("TJ_CONSOLIDATE_BASE_URL", server.url())
+            .env("TJ_EMBED", "hash")
+            .args(["consolidate"])
+            .assert()
+            .success()
+            .get_output()
+            .stdout
+            .clone()
+    };
+
+    let first = String::from_utf8(run()).unwrap();
+    assert!(
+        first.contains("consolidated 2 new fact(s)"),
+        "first run must store 2 facts; got: {first:?}"
+    );
+    // Second run: same facts already present -> de-duped to 0.
+    let second = String::from_utf8(run()).unwrap();
+    assert!(
+        second.contains("consolidated 0 new fact(s)"),
+        "second run must de-dup; got: {second:?}"
+    );
+    mock.assert();
+
+    // The fact is now recallable.
+    let recall = String::from_utf8(
+        Command::cargo_bin("task-journal")
+            .unwrap()
+            .env("XDG_DATA_HOME", xdg.path())
+            .env("TJ_EMBED", "hash")
+            .args(["recall", "refund ledger idempotent", "--k", "3"])
+            .assert()
+            .success()
+            .get_output()
+            .stdout
+            .clone(),
+    )
+    .unwrap();
+    assert!(
+        recall.contains("ledger"),
+        "consolidated fact must surface in cross-project recall; got: {recall:?}"
+    );
+}
+
+#[test]
+fn consolidate_skips_without_api_key_and_spends_nothing() {
+    // Safety: with no ANTHROPIC_API_KEY, consolidate makes no call and creates
+    // no facts — it can never spend automatically.
+    let xdg = assert_fs::TempDir::new().unwrap();
+    let proj = assert_fs::TempDir::new().unwrap();
+    let tid = String::from_utf8(
+        Command::cargo_bin("task-journal")
+            .unwrap()
+            .current_dir(proj.path())
+            .env("XDG_DATA_HOME", xdg.path())
+            .args(["create", "Scheduler"])
+            .assert()
+            .success()
+            .get_output()
+            .stdout
+            .clone(),
+    )
+    .unwrap()
+    .trim()
+    .to_string();
+    Command::cargo_bin("task-journal")
+        .unwrap()
+        .current_dir(proj.path())
+        .env("XDG_DATA_HOME", xdg.path())
+        .args([
+            "event",
+            &tid,
+            "--type",
+            "decision",
+            "--text",
+            "use postgres advisory locks for cron",
+        ])
+        .assert()
+        .success();
+
+    Command::cargo_bin("task-journal")
+        .unwrap()
+        .current_dir(proj.path())
+        .env("XDG_DATA_HOME", xdg.path())
+        .env_remove("ANTHROPIC_API_KEY")
+        .args(["consolidate"])
+        .assert()
+        .success()
+        .stdout(contains("skipped"));
+}