diff --git a/openspec/changes/promotion-candidates/design.md b/openspec/changes/promotion-candidates/design.md new file mode 100644 index 0000000..0b6d74e --- /dev/null +++ b/openspec/changes/promotion-candidates/design.md @@ -0,0 +1,41 @@ +## Where the detector runs + +Inside `run-once`, after jobs drain and before memory reconciliation, while the queue lock is held. The worker is the only writer of `lessons.jsonl` and `pending/`, so the lock serializes it. A detector error is logged and recorded in `status.json`'s existing `lastError`. It never fails the run, because memory delivery matters more than a proposal. + +## What a "lesson" is + +The worker's job carries one final assistant message, not structured lessons. The detector takes lines prefixed `LESSON:`, `GOTCHA:`, `PATTERN:` or `LEARNED:` (case-insensitive, list bullets allowed). When there are none, it takes the whole message (first 2,000 characters). It also takes typed `Lesson`/`Gotcha` entries from the project KB, because frontmatter tags are the only place a `global` tag is observable. Inline `#tag` words count as tags for session lessons. + +**The uncomfortable case:** a whole-message lesson is noisy. Two long, mostly different messages rarely reach Jaccard 0.6, which keeps false recurrence low. But a message that names a CLI in a code span and cites no repo path becomes a `portable-tooling` candidate on its own. The human gate absorbs that noise. If it proves too much, the fix is to tighten `names_dependency_or_cli`, not to remove the gate. + +## Fingerprints and ids + +- **Tokens:** lowercase alphanumeric runs. **Shingles:** word trigrams, hashed with the first 8 bytes of SHA-256. A text under three tokens is one shingle. SHA-256, not `DefaultHasher`, because the index outlives the binary and `DefaultHasher` output is not stable across Rust releases. +- **Lesson id:** SHA-256 over `projectId \0 source \0 normalized tokens`, so re-indexing the same job or KB entry is a no-op. +- **Clustering:** greedy, single pass, ordered by `(recordedAt, lessonId)`. Each record joins the first cluster whose anchor (first member) it matches at ≥ 0.6. New records always sort after existing ones, so an anchor never changes. +- **Candidate id:** `promo-` plus 16 hex digits of SHA-256 over the anchor's lesson id. It stays stable as evidence grows. + +## Idempotency + +A candidate whose id exists in `accepted/` or `rejected/` is skipped. A pending candidate is rewritten, atomically and keeping `createdAt`, only when its evidence set changed. Otherwise the file is untouched. + +**Known race:** `pk candidates accept` does not take the worker's queue lock, because pk-cli has no `fs2` dependency. If a worker run rewrites a pending candidate at the moment `accept` renames it, the pending file can reappear. A second accept is then idempotent: the KB upsert is skipped when the content is unchanged, and the memory operation is skipped when its id exists in any outbox state. The window is the time between one `exists` check and one `rename`. + +## Accept semantics + +The steps run in a crash-safe order: + +1. Upsert the KB entry. It has a fixed id and is skipped when unchanged. +2. Commit the shared snapshot. +3. Queue the memory operation. Its id is deterministic, `sha256("promotion-candidate:")`, and it is skipped when present in any `memory/*` state. +4. Rewrite the candidate with `state: accepted`, then durably rename it to `accepted/`. + +A crash at any step leaves a pending candidate, and re-running `accept` completes it without duplicates. + +The outbox record matches the worker's own `enqueue_memory`: schemaVersion 2, `method: add_memory`, and `payloadHash` = SHA-256 of `serde_json::to_vec(arguments)`. The worker's `normalize_operation` re-derives that hash, and it passes `add_memory` arguments through unchanged. + +**User scope** is `@user:` with `agent_id: @user` (design table §2). The hash uses the same resolver as the skill system's `project_id.py`: `PROMETHEUS_USER_ID`, else the first 16 hex digits of SHA-256 over the lowercased global git email. + +## Timestamps in pk-cli + +pk-cli has no chrono dependency, and adding one would change `Cargo.lock`, which the `--locked` rule forbids without approval. `rfc3339_now()` formats `SystemTime` with the civil-from-days algorithm. The worker parses `queuedAt` as RFC 3339, so the format must stay `YYYY-MM-DDTHH:MM:SSZ`. diff --git a/openspec/changes/promotion-candidates/proposal.md b/openspec/changes/promotion-candidates/proposal.md new file mode 100644 index 0000000..75f89d4 --- /dev/null +++ b/openspec/changes/promotion-candidates/proposal.md @@ -0,0 +1,44 @@ +## Why + +The skill system's team-aware learning design (§3, "Visibility and promotion") makes promotion beyond a project auto-proposed and human-confirmed. Nothing proposed or applied promotion candidates: + +- **No detector.** A lesson learned in two projects stayed two private project lessons. Nothing noticed recurrence, a `global` tag, or a lesson about a tool rather than a repository. +- **No confirmation step.** No command took a proposed lesson into the shared KB and the `@global`/`@user` memory scopes, so the only way to promote was a hand-run `pk ingest --scope shared`, which skips surreal-memory entirely. + +Adding a `pk` subcommand changes a frozen surface (the `pk` CLI). This proposal names the consumers. + +## What Changes + +- **Worker promotion detector (new, additive).** After draining jobs in `run-once`, `prometheus-learning-worker`: + - extracts lessons from each processed job (lines prefixed `LESSON:`, `GOTCHA:`, `PATTERN:` or `LEARNED:`, else the whole final assistant message) and from typed `Lesson`/`Gotcha` entries in the project KB; + - fingerprints them as word-trigram shingle sets into `~/.prometheus/learning-index/lessons.jsonl`, with project id, project root and source evidence. Append-only and idempotent on lesson id; + - proposes a candidate when a lesson recurs with Jaccard ≥ 0.6 across ≥ 2 distinct projects, is tagged global without a `[GLOBAL]`/`[USER]` marker, or names a dependency or CLI and contains no repo-relative path; + - writes candidates atomically to `~/.prometheus/promotion-candidates/pending/.json` with an evidence list. Re-runs do not duplicate a candidate, and a candidate already accepted or rejected is never re-proposed. +- **`pk candidates list|accept|reject --kind promotion|skill` (new subcommand, additive).** + - `list [--state pending|accepted|rejected] [--json]`. + - `accept ` (promotion): upserts the lesson into `~/.prometheus/knowledge/shared` as `promoted-`, commits the shared prompt snapshot, queues an `add_memory` operation for `@global` (or `@user:` / `@user` for a user-scoped candidate) in the learning-queue memory outbox, and moves the file to `accepted/`. + - `reject [--reason]`: moves the file to `rejected/`. + - `--kind skill` reads `~/.prometheus/skill-candidates/`. `list` and `reject` work. `accept` exits non-zero with "not yet supported" until skill promotion lands. + +## Consumers + +| Consumer | Affected surface | Effect | +|---|---|---| +| prometheus-skill-system (`kbd-open.sh`, reflector role, later skill-candidate work) | `pk candidates`, `~/.prometheus/promotion-candidates/` | Intended consumer. It surfaces pending candidates and asks a human to run `pk candidates accept|reject`. Needs a pin bump to this release; handoff note below. | +| prometheus-skills-mini | `pk ingest` | Unchanged. | +| forge-rs | `pk ingest` (shells out) | Unchanged. | +| prometheus-cli | pk-core / pk-store / pk-librarian crates | No public API change. | +| prometheus doctor | `status.json` | No field added. `lastError` may now carry a `promotion detector: …` message. | +| surreal-memory-server | `/api/v2/operations` via the worker's outbox | New operations use the existing `add_memory` method and the existing schemaVersion 2 outbox format, keyed `@global`/`@global` or `@user:`/`@user`. No ledger contract change. | + +## Capabilities + +### New Capabilities +- `promotion-candidates`: lesson fingerprinting, candidate proposal, and human-confirmed promotion to shared scopes. + +## Impact + +- **Code:** `pk-learning-worker/src/promotion.rs` (new), `pk-learning-worker/src/main.rs`, `pk-cli/src/candidates.rs` (new), `pk-cli/src/main.rs`. +- **Tests:** `pk-learning-worker/tests/promotion.rs`, `pk-cli/tests/candidates.rs`. +- **Dependencies:** none added. `Cargo.lock` is unchanged. +- **Downstream handoff (prometheus-skill-system):** bump the pk pin to the release containing this change. Document `pk candidates` in `docs/guide/13-tools-reference.md`, and point `kbd-open.sh` and the reflector role at `~/.prometheus/promotion-candidates/pending`. diff --git a/openspec/changes/promotion-candidates/specs/promotion-candidates/spec.md b/openspec/changes/promotion-candidates/specs/promotion-candidates/spec.md new file mode 100644 index 0000000..64dbe6f --- /dev/null +++ b/openspec/changes/promotion-candidates/specs/promotion-candidates/spec.md @@ -0,0 +1,43 @@ +## ADDED Requirements + +### Requirement: The worker fingerprints lessons +After processing jobs, `prometheus-learning-worker run-once` SHALL append a record for each newly seen lesson to `~/.prometheus/learning-index/lessons.jsonl`, carrying its project id, project root, source evidence and word-trigram fingerprint. Re-processing the same lesson SHALL NOT append a second record. + +#### Scenario: Re-running indexes nothing new +- **WHEN** the worker runs again with no new jobs +- **THEN** `lessons.jsonl` is byte-identical + +### Requirement: Recurring or portable lessons become promotion candidates +The worker SHALL write a candidate to `~/.prometheus/promotion-candidates/pending/.json`, atomically, with an evidence list, when a lesson recurs with Jaccard ≥ 0.6 across at least two distinct projects, is tagged global without a `[GLOBAL]`/`[USER]` marker, or names a dependency or CLI and contains no repo-relative path. A lesson carrying an immediate-promotion marker SHALL NOT be proposed. + +#### Scenario: The same lesson in two projects +- **WHEN** two jobs from two different projects carry the same lesson +- **THEN** exactly one pending candidate exists, with reason `recurrence` and two evidence entries citing two distinct project ids + +### Requirement: Candidate proposal is idempotent +Re-running the worker SHALL NOT create a second candidate for the same cluster, SHALL NOT rewrite a pending candidate whose evidence is unchanged, and SHALL NOT re-propose a candidate that was accepted or rejected. + +#### Scenario: Re-run adds none +- **WHEN** the worker runs a second time with no new jobs +- **THEN** the pending directory holds the same single file, byte-identical + +### Requirement: Accepting a promotion candidate promotes it +`pk candidates accept ` SHALL upsert the lesson into `~/.prometheus/knowledge/shared`, commit the shared prompt snapshot, queue one schemaVersion 2 `add_memory` operation whose `payloadHash` is the SHA-256 of its serialized arguments, keyed `@global`/`@global` (or `@user:`/`@user` for a user-scoped candidate), and move the candidate file to `accepted/`. Accepting an already-accepted candidate SHALL fail without queuing a second operation. + +#### Scenario: Accept a global candidate +- **WHEN** `pk candidates accept ` runs against a pending global candidate +- **THEN** the shared prompt snapshot contains `promoted-`, the learning-queue outbox holds one `@global` `add_memory` operation, and the candidate is in `accepted/` + +### Requirement: Rejecting a candidate retires it +`pk candidates reject ` SHALL move the candidate to `rejected/`, recording an optional reason. + +#### Scenario: Reject +- **WHEN** `pk candidates reject --reason noise` runs +- **THEN** the file is in `rejected/` and no longer in `pending/` + +### Requirement: Skill candidates are listed but not yet accepted +`pk candidates --kind skill` SHALL list and reject entries under `~/.prometheus/skill-candidates/`, and `accept` SHALL exit non-zero with a "not yet supported" message, leaving the file in place. + +#### Scenario: Skill accept +- **WHEN** `pk candidates accept --kind skill` runs +- **THEN** it exits non-zero, stderr says "not yet supported", and the candidate is still pending diff --git a/openspec/changes/promotion-candidates/tasks.md b/openspec/changes/promotion-candidates/tasks.md new file mode 100644 index 0000000..7aa16a2 --- /dev/null +++ b/openspec/changes/promotion-candidates/tasks.md @@ -0,0 +1,12 @@ +## 1. Promotion detector (learning-worker-engineer) +- [x] 1.1 `pk-learning-worker/src/promotion.rs`: lesson extraction, trigram fingerprints, `lessons.jsonl` index, greedy clustering, candidate proposal with evidence. +- [x] 1.2 Return observations from `process_job` and run the detector after the job loop in `run_once` (`pk-learning-worker/src/main.rs`). +- [x] 1.3 `pk-learning-worker/tests/promotion.rs`: two projects with one lesson yield one candidate with two evidence entries, and a re-run changes nothing. + +## 2. `pk candidates` (surfaces-engineer) +- [x] 2.1 `pk-cli/src/candidates.rs` + the `Candidates` subcommand in `pk-cli/src/main.rs`: list, accept (promotion), reject, `--kind skill` list/reject with accept refused. +- [x] 2.2 `pk-cli/tests/candidates.rs`: accept writes the shared snapshot, queues one `@global` op with a valid payloadHash, and moves the file; reject moves the file; skill accept exits non-zero. + +## 3. Release and handoff (knowledge-lead) +- [ ] 3.1 Verifier gate: `cargo fmt --all -- --check`, `cargo clippy --workspace --all-targets --locked -- -D warnings`, `cargo test --workspace --locked`. +- [ ] 3.2 Downstream handoff note for prometheus-skill-system (pin, tools reference, `kbd-open.sh`/reflector surfacing). diff --git a/pk-cli/src/candidates.rs b/pk-cli/src/candidates.rs new file mode 100644 index 0000000..c3450b1 --- /dev/null +++ b/pk-cli/src/candidates.rs @@ -0,0 +1,473 @@ +//! `pk candidates list|accept|reject` — the human confirmation step for +//! promotion candidates proposed by the learning worker (design: +//! team-aware-learning-memory §3). Skill candidates share the directory +//! layout; accepting them is not implemented yet. + +use anyhow::{bail, Context, Result}; +use clap::{Subcommand, ValueEnum}; +use pk_core::{paths::home_dir, ArticleId, WikiEntry}; +use pk_store::{commit_prompt_snapshot, MarkdownStore}; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; +use std::{ + fs::{self, File, OpenOptions}, + io::Write, + path::{Path, PathBuf}, + process::Command, + sync::atomic::{AtomicU64, Ordering}, + time::{SystemTime, UNIX_EPOCH}, +}; + +static TEMPORARY_SEQUENCE: AtomicU64 = AtomicU64::new(0); +const MEMORY_STATES: &[&str] = &[ + "pending", + "submitting", + "accepted", + "retry", + "completed", + "rejected", + "dead-letter", + "stalled", +]; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)] +pub enum CandidateKind { + /// Lessons proposed for user/global scope (`~/.prometheus/promotion-candidates`). + Promotion, + /// Proposed skills (`~/.prometheus/skill-candidates`). + Skill, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)] +pub enum CandidateState { + Pending, + Accepted, + Rejected, +} + +impl CandidateState { + fn dir(self) -> &'static str { + match self { + Self::Pending => "pending", + Self::Accepted => "accepted", + Self::Rejected => "rejected", + } + } +} + +#[derive(Debug, Subcommand)] +pub enum CandidatesCmd { + /// List candidates (pending by default). + List { + #[arg(long, value_enum, default_value = "pending")] + state: CandidateState, + #[arg(long, default_value_t = false)] + json: bool, + }, + /// Accept a pending candidate: write it to the shared KB, publish the shared + /// prompt snapshot, queue a surreal-memory operation, move it to accepted/. + Accept { + #[arg()] + id: String, + }, + /// Reject a pending candidate: move it to rejected/. It is never proposed again. + Reject { + #[arg()] + id: String, + #[arg(long)] + reason: Option, + }, +} + +pub async fn run(kind: CandidateKind, action: CandidatesCmd) -> Result<()> { + let home = home_dir().context("HOME unavailable")?; + let root = candidates_root(&home, kind); + match action { + CandidatesCmd::List { state, json } => list(&root, state, json), + CandidatesCmd::Accept { id } => match kind { + CandidateKind::Promotion => accept_promotion(&home, &root, &id).await, + CandidateKind::Skill => bail!( + "accepting skill candidates is not yet supported; use `pk candidates reject --kind skill` or wait for skill promotion support" + ), + }, + CandidatesCmd::Reject { id, reason } => reject(&root, &id, reason.as_deref()), + } +} + +fn candidates_root(home: &Path, kind: CandidateKind) -> PathBuf { + home.join(".prometheus").join(match kind { + CandidateKind::Promotion => "promotion-candidates", + CandidateKind::Skill => "skill-candidates", + }) +} + +fn list(root: &Path, state: CandidateState, json_output: bool) -> Result<()> { + let directory = root.join(state.dir()); + let mut paths: Vec = match fs::read_dir(&directory) { + Ok(entries) => entries + .filter_map(std::result::Result::ok) + .map(|entry| entry.path()) + .filter(|path| path.extension().and_then(|e| e.to_str()) == Some("json")) + .collect(), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Vec::new(), + Err(error) => return Err(error.into()), + }; + paths.sort(); + let mut candidates = Vec::new(); + for path in paths { + let mut value: Value = serde_json::from_slice(&fs::read(&path)?) + .with_context(|| format!("unreadable candidate {}", path.display()))?; + if value.get("id").and_then(Value::as_str).is_none() { + if let Some(stem) = path.file_stem().and_then(|s| s.to_str()) { + value["id"] = Value::String(stem.to_owned()); + } + } + candidates.push(value); + } + if json_output { + println!("{}", serde_json::to_string_pretty(&candidates)?); + return Ok(()); + } + if candidates.is_empty() { + println!("no {} candidates", state.dir()); + return Ok(()); + } + for candidate in &candidates { + let text = |key: &str| candidate.get(key).and_then(Value::as_str).unwrap_or("-"); + let reasons = candidate + .get("reasons") + .and_then(Value::as_array) + .map(|items| { + items + .iter() + .filter_map(Value::as_str) + .collect::>() + .join(",") + }) + .unwrap_or_else(|| "-".to_owned()); + let evidence = candidate + .get("evidence") + .and_then(Value::as_array) + .map_or(0, Vec::len); + let title = candidate + .get("title") + .or_else(|| candidate.get("name")) + .and_then(Value::as_str) + .unwrap_or("-"); + println!( + "{}\t{}\t{}\tevidence={}\t{}", + text("id"), + text("scope"), + reasons, + evidence, + title + ); + } + Ok(()) +} + +/// Candidate ids become file names, so only a plain file-name alphabet is accepted. +fn validate_id(id: &str) -> Result<()> { + let safe = !id.is_empty() + && !id.starts_with('.') + && id + .chars() + .all(|c| c.is_ascii_alphanumeric() || "-_.".contains(c)); + if !safe { + bail!("invalid candidate id {id:?}"); + } + Ok(()) +} + +fn pending_candidate(root: &Path, id: &str) -> Result<(PathBuf, Value)> { + validate_id(id)?; + let filename = format!("{id}.json"); + let pending = root.join("pending").join(&filename); + if !pending.exists() { + for decided in ["accepted", "rejected"] { + if root.join(decided).join(&filename).exists() { + bail!("candidate {id} is already {decided}"); + } + } + bail!("no pending candidate {id} under {}", root.display()); + } + let value = serde_json::from_slice(&fs::read(&pending)?) + .with_context(|| format!("unreadable candidate {}", pending.display()))?; + Ok((pending, value)) +} + +async fn accept_promotion(home: &Path, root: &Path, id: &str) -> Result<()> { + let (pending, mut candidate) = pending_candidate(root, id)?; + let content = candidate + .get("content") + .and_then(Value::as_str) + .filter(|text| !text.trim().is_empty()) + .context("candidate has no content")? + .to_owned(); + let title = candidate + .get("title") + .and_then(Value::as_str) + .unwrap_or("Promoted lesson") + .to_owned(); + let scope = match candidate.get("scope").and_then(Value::as_str) { + Some("user") => "user", + _ => "global", + }; + + let entry_id = upsert_shared(home, id, &title, &content, scope, &candidate).await?; + let operation_id = queue_memory(home, id, &content, scope)?; + + let now = rfc3339_now(); + candidate["state"] = json!("accepted"); + candidate["acceptedAt"] = json!(now); + candidate["updatedAt"] = json!(now); + candidate["sharedEntryId"] = json!(entry_id); + candidate["memoryOperationId"] = json!(operation_id); + atomic_json(&pending, &candidate)?; + let accepted = root.join("accepted").join(format!("{id}.json")); + durable_rename(&pending, &accepted)?; + println!("accepted {id} → shared entry {entry_id}, memory operation {operation_id}"); + Ok(()) +} + +/// Write the lesson into `~/.prometheus/knowledge/shared` and publish the shared +/// prompt snapshot that `pk context` reads. Idempotent on retry. +async fn upsert_shared( + home: &Path, + id: &str, + title: &str, + content: &str, + scope: &str, + candidate: &Value, +) -> Result { + let shared = home.join(".prometheus/knowledge/shared"); + let store = MarkdownStore::open(&shared).await?; + let entry_id = format!("promoted-{id}"); + let article_id = ArticleId::from(entry_id.clone()); + let unchanged = store + .get(&article_id) + .await + .is_ok_and(|existing| existing.content == content); + if !unchanged { + let mut entry = WikiEntry::new(title.to_owned(), content.to_owned()); + entry.id = article_id.clone(); + entry.entry_type = Some("Lesson".to_owned()); + let mut tags: Vec = candidate + .get("tags") + .and_then(Value::as_array) + .map(|items| { + items + .iter() + .filter_map(Value::as_str) + .map(str::to_owned) + .collect() + }) + .unwrap_or_default(); + for tag in ["promoted".to_owned(), format!("scope:{scope}")] { + if !tags.contains(&tag) { + tags.push(tag); + } + } + entry.tags = tags; + entry.sources = candidate + .get("evidence") + .and_then(Value::as_array) + .map(|items| { + items + .iter() + .filter_map(|item| { + let project = item.get("projectId").and_then(Value::as_str)?; + let source = item.get("source").and_then(Value::as_str)?; + Some(format!("{project}:{source}").into()) + }) + .collect() + }) + .unwrap_or_default(); + store.upsert(entry).await?; + store.regenerate_index().await?; + store.append_log("Promote", title, &article_id).await?; + } + commit_prompt_snapshot(&shared, "shared", store.snapshot().await?)?; + Ok(entry_id) +} + +/// Queue an `add_memory` operation in the learning-queue outbox, in the +/// worker's schemaVersion 2 format: `payloadHash` is the SHA-256 of the +/// serialized `arguments`, which the worker re-checks before submitting. +fn queue_memory(home: &Path, id: &str, content: &str, scope: &str) -> Result { + let queue = std::env::var_os("PROMETHEUS_LEARNING_QUEUE") + .filter(|value| !value.is_empty()) + .map(PathBuf::from) + .unwrap_or_else(|| home.join(".prometheus/learning-queue")); + let operation_id = hex_digest(format!("promotion-candidate:{id}").as_bytes()); + let filename = format!("{operation_id}.json"); + if MEMORY_STATES + .iter() + .any(|state| queue.join("memory").join(state).join(&filename).exists()) + { + return Ok(operation_id); + } + let (user_id, agent_id) = match scope { + "user" => (resolve_user_scope(), "@user".to_owned()), + _ => ("@global".to_owned(), "@global".to_owned()), + }; + let arguments = json!({ + "content": content, + "user_id": user_id, + "agent_id": agent_id, + "session_id": format!("promotion:{id}"), + "categories": ["promotion", format!("scope:{scope}")] + }); + let payload_hash = hex_digest(&serde_json::to_vec(&arguments)?); + let operation = json!({ + "schemaVersion": 2, + "operationId": operation_id, + "method": "add_memory", + "arguments": arguments, + "dependencies": [], + "payloadHash": payload_hash, + "state": "pending", + "queuedAt": rfc3339_now(), + "lastError": null, + "receipt": null + }); + let pending = queue.join("memory/pending"); + fs::create_dir_all(&pending)?; + harden_directory(&pending)?; + atomic_json(&pending.join(&filename), &operation)?; + Ok(operation_id) +} + +/// `@user:`: `PROMETHEUS_USER_ID` when set, else the first 16 hex digits +/// of SHA-256 over the lowercased global git email, matching the skill +/// system's `project_id.py` resolver. +fn resolve_user_scope() -> String { + if let Ok(value) = std::env::var("PROMETHEUS_USER_ID") { + let value = value.trim(); + if !value.is_empty() { + let bare = value + .strip_prefix("@user:") + .or_else(|| value.strip_prefix("user:")) + .unwrap_or(value); + return format!("@user:{bare}"); + } + } + let email = Command::new("git") + .args(["config", "--global", "user.email"]) + .output() + .ok() + .filter(|output| output.status.success()) + .map(|output| { + String::from_utf8_lossy(&output.stdout) + .trim() + .to_lowercase() + }) + .filter(|email| !email.is_empty()); + match email { + Some(email) => format!("@user:{}", &hex_digest(email.as_bytes())[..16]), + None => "@user:unknown".to_owned(), + } +} + +fn reject(root: &Path, id: &str, reason: Option<&str>) -> Result<()> { + let (pending, mut candidate) = pending_candidate(root, id)?; + let now = rfc3339_now(); + candidate["state"] = json!("rejected"); + candidate["rejectedAt"] = json!(now); + candidate["updatedAt"] = json!(now); + if let Some(reason) = reason { + candidate["rejectionReason"] = json!(reason); + } + atomic_json(&pending, &candidate)?; + durable_rename(&pending, &root.join("rejected").join(format!("{id}.json")))?; + println!("rejected {id}"); + Ok(()) +} + +fn hex_digest(bytes: &[u8]) -> String { + Sha256::digest(bytes) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect() +} + +fn atomic_json(path: &Path, value: &Value) -> Result<()> { + let parent = path.parent().context("output path has no parent")?; + fs::create_dir_all(parent)?; + let temporary = parent.join(format!( + ".{}.{}.{}.tmp", + path.file_name() + .context("output path has no filename")? + .to_string_lossy(), + std::process::id(), + TEMPORARY_SEQUENCE.fetch_add(1, Ordering::Relaxed) + )); + let mut file = OpenOptions::new() + .create_new(true) + .write(true) + .open(&temporary)?; + serde_json::to_writer_pretty(&mut file, value)?; + writeln!(file)?; + file.sync_all()?; + fs::rename(&temporary, path)?; + sync_directory(parent) +} + +fn durable_rename(source: &Path, target: &Path) -> Result<()> { + let target_parent = target.parent().context("target path has no parent")?; + fs::create_dir_all(target_parent)?; + fs::rename(source, target)?; + sync_directory(target_parent)?; + if let Some(source_parent) = source.parent().filter(|parent| *parent != target_parent) { + sync_directory(source_parent)?; + } + Ok(()) +} + +#[cfg(unix)] +fn sync_directory(path: &Path) -> Result<()> { + File::open(path)?.sync_all()?; + Ok(()) +} + +#[cfg(not(unix))] +fn sync_directory(_path: &Path) -> Result<()> { + Ok(()) +} + +#[cfg(unix)] +fn harden_directory(path: &Path) -> Result<()> { + use std::os::unix::fs::PermissionsExt; + fs::set_permissions(path, fs::Permissions::from_mode(0o700))?; + Ok(()) +} + +#[cfg(not(unix))] +fn harden_directory(_path: &Path) -> Result<()> { + Ok(()) +} + +/// UTC now as RFC 3339 (`YYYY-MM-DDTHH:MM:SSZ`). pk-cli has no chrono +/// dependency, and adding one would change Cargo.lock. +fn rfc3339_now() -> String { + let seconds = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |elapsed| elapsed.as_secs()) as i64; + let (days, time) = (seconds.div_euclid(86_400), seconds.rem_euclid(86_400)); + // Civil-from-days (Howard Hinnant), proleptic Gregorian calendar. + let z = days + 719_468; + let era = z.div_euclid(146_097); + let doe = z.rem_euclid(146_097); + let yoe = (doe - doe / 1_460 + doe / 36_524 - doe / 146_096) / 365; + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); + let mp = (5 * doy + 2) / 153; + let day = doy - (153 * mp + 2) / 5 + 1; + let month = if mp < 10 { mp + 3 } else { mp - 9 }; + let year = yoe + era * 400 + i64::from(month <= 2); + format!( + "{year:04}-{month:02}-{day:02}T{:02}:{:02}:{:02}Z", + time / 3_600, + (time % 3_600) / 60, + time % 60 + ) +} diff --git a/pk-cli/src/main.rs b/pk-cli/src/main.rs index f1cfea6..bbc0ea7 100644 --- a/pk-cli/src/main.rs +++ b/pk-cli/src/main.rs @@ -23,6 +23,8 @@ use std::{ }; use tokio::sync::broadcast; +mod candidates; + /// KB scope: project-local (default) or globally shared across projects. #[derive(Debug, Clone, Copy, ValueEnum)] enum KbScope { @@ -220,6 +222,14 @@ enum Cmd { #[command(subcommand)] action: EventsCmd, }, + /// Review promotion or skill candidates proposed by the learning worker + Candidates { + /// Candidate kind: promotion (lessons for user/global scope) or skill. + #[arg(long, value_enum, default_value = "promotion", global = true)] + kind: candidates::CandidateKind, + #[command(subcommand)] + action: candidates::CandidatesCmd, + }, /// Migrate single-store events to dual-store (KG + episodic) layout (SP-020) MigrateStores { /// Print migration plan without applying changes (default: true) @@ -281,6 +291,11 @@ async fn main() -> Result<()> { .with_writer(std::io::stderr) .init(); + // Candidates live under ~/.prometheus and never touch the resolved KB. + if let Cmd::Candidates { kind, action } = cli.command { + return candidates::run(kind, action).await; + } + // Resolve KB directory: explicit flag > env > project-root > global fallback let kb_dir = resolve_kb_dir(cli.kb_dir.as_deref(), &cli.command); @@ -567,6 +582,7 @@ async fn main() -> Result<()> { } Cmd::MigrateToPerProject { .. } => unreachable!("handled above"), + Cmd::Candidates { .. } => unreachable!("candidates return before resolving the KB"), Cmd::Codegraph { action } => match action { CodegraphCmd::Extract { diff --git a/pk-cli/tests/candidates.rs b/pk-cli/tests/candidates.rs new file mode 100644 index 0000000..003a4fe --- /dev/null +++ b/pk-cli/tests/candidates.rs @@ -0,0 +1,168 @@ +//! `pk candidates accept` promotes a pending lesson: it lands in the shared KB's +//! committed prompt snapshot, an `@global` add_memory operation is queued in the +//! learning-queue outbox (schemaVersion 2, payloadHash over the arguments), and +//! the candidate file moves to accepted/. + +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; +use std::{ + fs, + path::{Path, PathBuf}, + process::{Command, Output}, +}; + +fn pk(home: &Path, args: &[&str]) -> Output { + Command::new(env!("CARGO_BIN_EXE_pk")) + .current_dir(home) + .env("HOME", home) + .env("RUST_LOG", "error") + .env_remove("PROMETHEUS_LEARNING_QUEUE") + .env_remove("PK_KB_DIR") + .args(args) + .output() + .unwrap() +} + +fn write_candidate(home: &Path, kind_dir: &str, id: &str, content: &str) -> PathBuf { + let path = home + .join(".prometheus") + .join(kind_dir) + .join("pending") + .join(format!("{id}.json")); + fs::create_dir_all(path.parent().unwrap()).unwrap(); + fs::write( + &path, + serde_json::to_vec_pretty(&json!({ + "schemaVersion": 1, + "id": id, + "kind": "promotion", + "state": "pending", + "scope": "global", + "reasons": ["recurrence"], + "title": "Anchor hook matchers to the plugin prefix", + "content": content, + "tags": ["hooks"], + "fingerprint": "f", + "evidence": [ + {"lessonId": "l1", "projectId": "project:alpha", "projectRoot": "/alpha", + "source": "session:s1#0", "recordedAt": "2026-10-04T00:00:00Z", "similarity": 1.0}, + {"lessonId": "l2", "projectId": "project:beta", "projectRoot": "/beta", + "source": "session:s2#0", "recordedAt": "2026-10-04T00:00:00Z", "similarity": 1.0} + ], + "createdAt": "2026-10-04T00:00:00Z", + "updatedAt": "2026-10-04T00:00:00Z" + })) + .unwrap(), + ) + .unwrap(); + path +} + +fn assert_ok(output: &Output, what: &str) { + assert!( + output.status.success(), + "{what} failed: {}", + String::from_utf8_lossy(&output.stderr) + ); +} + +#[test] +fn accept_writes_the_shared_snapshot_queues_a_global_operation_and_moves_the_file() { + let fixture = tempfile::tempdir().unwrap(); + let home = fixture.path().join("home"); + fs::create_dir_all(&home).unwrap(); + let id = "promo-0123456789abcdef"; + let content = "Anchor hook matchers to the plugin prefix so renamed agents still match."; + let pending = write_candidate(&home, "promotion-candidates", id, content); + + let listed = pk(&home, &["candidates", "list"]); + assert_ok(&listed, "candidates list"); + assert!(String::from_utf8_lossy(&listed.stdout).contains(id)); + + let accepted = pk(&home, &["candidates", "accept", id, "--kind", "promotion"]); + assert_ok(&accepted, "candidates accept"); + + // Shared KB entry, published through the committed prompt snapshot. + let shared = home.join(".prometheus/knowledge/shared"); + let snapshot = pk_store::read_prompt_snapshot(&shared, "shared") + .expect("accept must commit the shared prompt snapshot"); + let entry = snapshot + .entries + .iter() + .find(|entry| entry.id.as_str() == format!("promoted-{id}")) + .unwrap_or_else(|| panic!("promoted entry missing from shared snapshot")); + assert_eq!(entry.content.trim(), content); + + // Exactly one @global add_memory operation, hash-bound to its arguments. + let outbox = home.join(".prometheus/learning-queue/memory/pending"); + let operations: Vec = fs::read_dir(&outbox) + .unwrap() + .flatten() + .filter(|file| file.path().extension().and_then(|v| v.to_str()) == Some("json")) + .map(|file| serde_json::from_slice(&fs::read(file.path()).unwrap()).unwrap()) + .collect(); + assert_eq!(operations.len(), 1, "{operations:#?}"); + let operation = &operations[0]; + assert_eq!(operation["schemaVersion"], 2); + assert_eq!(operation["method"], "add_memory"); + assert_eq!( + operation["arguments"]["user_id"], "@global", + "{operation:#}" + ); + assert_eq!( + operation["arguments"]["agent_id"], "@global", + "{operation:#}" + ); + assert_eq!(operation["arguments"]["content"], content); + let expected_hash: String = + Sha256::digest(serde_json::to_vec(&operation["arguments"]).unwrap()) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect(); + assert_eq!(operation["payloadHash"], expected_hash.as_str()); + + // The candidate moved to accepted/. + assert!(!pending.exists(), "pending file must be gone"); + let moved = home.join(format!( + ".prometheus/promotion-candidates/accepted/{id}.json" + )); + let record: Value = serde_json::from_slice(&fs::read(&moved).unwrap()).unwrap(); + assert_eq!(record["state"], "accepted"); + + // A second accept is refused, not duplicated. + let again = pk(&home, &["candidates", "accept", id]); + assert!(!again.status.success()); + assert_eq!(fs::read_dir(&outbox).unwrap().count(), 1); +} + +#[test] +fn reject_moves_the_file_and_skill_accept_is_not_yet_supported() { + let fixture = tempfile::tempdir().unwrap(); + let home = fixture.path().join("home"); + fs::create_dir_all(&home).unwrap(); + let pending = write_candidate(&home, "promotion-candidates", "promo-reject", "A lesson."); + let rejected = pk( + &home, + &["candidates", "reject", "promo-reject", "--reason", "noise"], + ); + assert_ok(&rejected, "candidates reject"); + assert!(!pending.exists()); + assert!(home + .join(".prometheus/promotion-candidates/rejected/promo-reject.json") + .exists()); + + let skill = write_candidate(&home, "skill-candidates", "skill-one", "A skill."); + let listed = pk(&home, &["candidates", "list", "--kind", "skill"]); + assert_ok(&listed, "candidates list --kind skill"); + assert!(String::from_utf8_lossy(&listed.stdout).contains("skill-one")); + let accept = pk( + &home, + &["candidates", "accept", "skill-one", "--kind", "skill"], + ); + assert!(!accept.status.success(), "skill accept must exit non-zero"); + assert!(String::from_utf8_lossy(&accept.stderr).contains("not yet supported")); + assert!( + skill.exists(), + "an unsupported accept leaves the file in place" + ); +} diff --git a/pk-learning-worker/src/main.rs b/pk-learning-worker/src/main.rs index 801855b..a0d3fe0 100644 --- a/pk-learning-worker/src/main.rs +++ b/pk-learning-worker/src/main.rs @@ -18,6 +18,8 @@ use std::{ time::{Duration, Instant}, }; +mod promotion; + static TEMPORARY_SEQUENCE: AtomicU64 = AtomicU64::new(0); const MEMORY_REQUEST_TIMEOUT: Duration = Duration::from_secs(30); /// First wait before re-polling an accepted operation whose receipt did not change. @@ -366,9 +368,13 @@ async fn run_once(root: &Path, memory_url: &str, stale_after: Duration) -> Resul started_at: Utc::now().to_rfc3339(), ..RunSummary::default() }; + let mut observations = Vec::new(); for path in json_files(&root.join("pending"))? { match process_job(root, &path).await { - Ok(()) => summary.jobs_completed += 1, + Ok(observed) => { + summary.jobs_completed += 1; + observations.extend(observed); + } Err(error) => { summary.last_error = Some(error.to_string()); let completed = root @@ -386,6 +392,11 @@ async fn run_once(root: &Path, memory_url: &str, stale_after: Duration) -> Resul } } } + // Promotion only proposes; a failure here must not hold up memory delivery. + if let Err(error) = promotion::run(&observations) { + tracing::warn!(error = %error, "promotion detector failed"); + summary.last_error = Some(format!("promotion detector: {error}")); + } let client = memory_client(MEMORY_REQUEST_TIMEOUT)?; // A record reconciled from `memory/submitting` lands in `memory/accepted` @@ -511,7 +522,7 @@ fn migrate_legacy_memory_retry(root: &Path) -> Result<()> { Ok(()) } -async fn process_job(root: &Path, pending_path: &Path) -> Result<()> { +async fn process_job(root: &Path, pending_path: &Path) -> Result> { let name = pending_path .file_name() .context("job path has no filename")?; @@ -525,7 +536,7 @@ async fn process_job(root: &Path, pending_path: &Path) -> Result<()> { let completed = root.join("completed").join(name); if completed.exists() { preserve_duplicate(root, &processing, "duplicate-completed")?; - return Ok(()); + return Ok(Vec::new()); } let packet = build_session_packet(&job)?; @@ -569,18 +580,51 @@ async fn process_job(root: &Path, pending_path: &Path) -> Result<()> { // Publish the store's current entries as the committed prompt snapshot that // `pk context` reads. Without this, session records were written to disk // but never became recallable. + let entries = store.snapshot().await?; + let project_id = lesson_project_id(&job); + let mut observations = job + .transcript_path + .as_deref() + .and_then(extract_final_assistant_message) + .map(|message| { + promotion::observations_from_message( + &project_id, + &job.project_root, + &job.session_id, + &message, + ) + }) + .unwrap_or_default(); + if matches!(job.scope, LearningScope::Project) { + observations.extend(promotion::observations_from_entries( + &project_id, + &job.project_root, + &entries, + )); + } pk_store::commit_prompt_snapshot( &target_kb, match job.scope { LearningScope::Project => "project", LearningScope::Shared => "shared", }, - store.snapshot().await?, + entries, )?; append_learning_log(&job, &packet)?; enqueue_memory(root, &job, &packet)?; durable_rename(&processing, &completed)?; - Ok(()) + Ok(observations) +} + +/// The project a lesson belongs to, for cross-project recurrence: the job's +/// resolved project id, else the same fallback the memory scope keys use. +fn lesson_project_id(job: &LearningJob) -> String { + job.project_id + .as_deref() + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_owned) + .unwrap_or_else(|| project_scope(&job.project_root)) } fn build_session_packet(job: &LearningJob) -> Result { diff --git a/pk-learning-worker/src/promotion.rs b/pk-learning-worker/src/promotion.rs new file mode 100644 index 0000000..36e00ec --- /dev/null +++ b/pk-learning-worker/src/promotion.rs @@ -0,0 +1,627 @@ +//! Promotion detector (design: team-aware-learning-memory §3). +//! +//! After each `run-once`, lessons seen by the worker are fingerprinted into +//! `~/.prometheus/learning-index/lessons.jsonl`, and lessons that look reusable +//! beyond their project become promotion candidates under +//! `~/.prometheus/promotion-candidates/pending/.json`. A human accepts or +//! rejects a candidate with `pk candidates`; nothing here writes a shared scope. +//! +//! A lesson becomes a candidate when any of these holds: +//! - it recurs (word-trigram Jaccard >= 0.6) in at least two distinct projects; +//! - it is tagged global but carries no `[GLOBAL]` marker (marker lines are +//! promoted immediately by another path, so they are never proposed); +//! - it names a dependency or CLI and contains no repo-relative path. + +use anyhow::{Context, Result}; +use chrono::Utc; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use sha2::{Digest, Sha256}; +use std::{ + collections::{BTreeSet, HashSet}, + fs::{self, File, OpenOptions}, + io::{BufRead, BufReader, Write}, + path::{Path, PathBuf}, +}; + +use crate::{atomic_json, harden_directory}; + +pub(crate) const RECURRENCE_THRESHOLD: f64 = 0.6; +const SHINGLE_SIZE: usize = 3; +const LESSON_TEXT_LIMIT: usize = 2_000; +const LESSON_PREFIXES: &[&str] = &["lesson:", "gotcha:", "pattern:", "learned:"]; +const IMMEDIATE_MARKERS: &[&str] = &["[GLOBAL]", "[USER]"]; +const GLOBAL_TAGS: &[&str] = &["global", "scope:global", "vis:global"]; +const USER_TAGS: &[&str] = &["user", "scope:user", "vis:user"]; +/// Command-line tools whose name inside a code span counts as "names a CLI". +const KNOWN_CLIS: &[&str] = &[ + "apt", + "brew", + "bun", + "bunx", + "cargo", + "claude", + "cmake", + "codex", + "curl", + "deno", + "docker", + "gh", + "git", + "go", + "gradle", + "helm", + "jq", + "kubectl", + "launchctl", + "make", + "mvn", + "node", + "npm", + "npx", + "pip", + "pip3", + "pk", + "pnpm", + "python", + "python3", + "rustc", + "rustup", + "sqlite3", + "systemctl", + "terraform", + "uv", + "uvx", + "yarn", +]; +/// First path segments that mark a repo-relative path even without an extension. +const REPO_DIRS: &[&str] = &[ + ".claude", + ".github", + ".kbd-orchestrator", + ".prometheus", + "crates", + "docs", + "hooks", + "lib", + "scripts", + "shared", + "skills", + "src", + "tests", +]; + +/// A lesson observed during this run, before fingerprinting. +#[derive(Debug, Clone)] +pub(crate) struct Observation { + pub project_id: String, + pub project_root: PathBuf, + /// Where the lesson came from: `session:#` or `kb:`. + pub source: String, + pub session_id: Option, + pub text: String, + pub tags: Vec, +} + +/// One line of `lessons.jsonl`. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +struct LessonRecord { + schema_version: u32, + lesson_id: String, + project_id: String, + project_root: PathBuf, + source: String, + #[serde(default)] + session_id: Option, + recorded_at: String, + text: String, + #[serde(default)] + tags: Vec, + /// Lesson carries an immediate-promotion marker; never proposed. + #[serde(default)] + marked: bool, + /// Sorted, de-duplicated word-trigram hashes. + shingles: Vec, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +struct Evidence { + lesson_id: String, + project_id: String, + project_root: PathBuf, + source: String, + #[serde(default)] + session_id: Option, + recorded_at: String, + similarity: f64, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +struct Candidate { + schema_version: u32, + id: String, + kind: String, + state: String, + /// `global` or `user`. + scope: String, + reasons: Vec, + title: String, + content: String, + tags: Vec, + fingerprint: String, + evidence: Vec, + created_at: String, + updated_at: String, +} + +/// Turn the lessons of one processed job into observations. +pub(crate) fn observations_from_message( + project_id: &str, + project_root: &Path, + session_id: &str, + message: &str, +) -> Vec { + extract_lessons(message) + .into_iter() + .enumerate() + .map(|(index, text)| Observation { + project_id: project_id.to_owned(), + project_root: project_root.to_path_buf(), + source: format!("session:{session_id}#{index}"), + session_id: Some(session_id.to_owned()), + tags: hashtags(&text), + text, + }) + .collect() +} + +/// Typed lessons already in a project knowledge base (`Lesson`, `Gotcha`). +/// Their frontmatter tags are what makes "tagged global" observable. +pub(crate) fn observations_from_entries( + project_id: &str, + project_root: &Path, + entries: &[pk_core::WikiEntry], +) -> Vec { + entries + .iter() + .filter(|entry| { + entry.entry_type.as_deref().is_some_and(|kind| { + kind.eq_ignore_ascii_case("lesson") || kind.eq_ignore_ascii_case("gotcha") + }) + }) + .map(|entry| Observation { + project_id: project_id.to_owned(), + project_root: project_root.to_path_buf(), + source: format!("kb:{}", entry.id.as_str()), + session_id: None, + text: entry + .content + .trim() + .chars() + .take(LESSON_TEXT_LIMIT) + .collect(), + tags: entry + .tags + .iter() + .map(|tag| tag.to_ascii_lowercase()) + .collect(), + }) + .collect() +} + +/// Fingerprint new observations, then propose candidates from the whole index. +pub(crate) fn run(observations: &[Observation]) -> Result { + let home = pk_core::paths::home_dir().context("HOME unavailable")?; + run_in(&home.join(".prometheus"), observations) +} + +fn run_in(prometheus_home: &Path, observations: &[Observation]) -> Result { + let index_path = prometheus_home.join("learning-index/lessons.jsonl"); + let mut records = read_index(&index_path)?; + append_new(&index_path, &mut records, observations)?; + propose(&prometheus_home.join("promotion-candidates"), &records) +} + +fn read_index(path: &Path) -> Result> { + if !path.exists() { + return Ok(Vec::new()); + } + let mut records = Vec::new(); + for line in BufReader::new(File::open(path)?) + .lines() + .map_while(std::result::Result::ok) + { + // A torn final line from a crash mid-append is skipped, not fatal. + if let Ok(record) = serde_json::from_str::(&line) { + records.push(record); + } + } + Ok(records) +} + +fn append_new( + path: &Path, + records: &mut Vec, + observations: &[Observation], +) -> Result<()> { + let mut known: HashSet = records.iter().map(|r| r.lesson_id.clone()).collect(); + let now = Utc::now().to_rfc3339(); + let mut fresh = Vec::new(); + for observation in observations { + let tokens = tokens(&observation.text); + if tokens.is_empty() { + continue; + } + let lesson_id = hex_digest(&format!( + "{}\0{}\0{}", + observation.project_id, + observation.source, + tokens.join(" ") + ))[..32] + .to_owned(); + if !known.insert(lesson_id.clone()) { + continue; + } + fresh.push(LessonRecord { + schema_version: 1, + lesson_id, + project_id: observation.project_id.clone(), + project_root: observation.project_root.clone(), + source: observation.source.clone(), + session_id: observation.session_id.clone(), + recorded_at: now.clone(), + text: observation.text.chars().take(LESSON_TEXT_LIMIT).collect(), + tags: observation.tags.clone(), + marked: IMMEDIATE_MARKERS + .iter() + .any(|marker| observation.text.contains(marker)), + shingles: shingles(&tokens), + }); + } + if fresh.is_empty() { + return Ok(()); + } + let parent = path.parent().context("index path has no parent")?; + fs::create_dir_all(parent)?; + harden_directory(parent)?; + let mut file = OpenOptions::new().create(true).append(true).open(path)?; + for record in &fresh { + serde_json::to_writer(&mut file, record)?; + writeln!(file)?; + } + file.sync_data()?; + records.extend(fresh); + Ok(()) +} + +/// Greedy single-pass clustering in record order, so an existing cluster's +/// anchor (and therefore its candidate id) never changes as the index grows. +fn propose(candidates_root: &Path, records: &[LessonRecord]) -> Result { + let mut ordered: Vec<&LessonRecord> = records.iter().filter(|r| !r.marked).collect(); + ordered.sort_by(|a, b| { + a.recorded_at + .cmp(&b.recorded_at) + .then_with(|| a.lesson_id.cmp(&b.lesson_id)) + }); + let mut clusters: Vec> = Vec::new(); + for record in ordered { + let joined = clusters.iter_mut().find_map(|cluster| { + let similarity = jaccard(&cluster[0].0.shingles, &record.shingles); + (similarity >= RECURRENCE_THRESHOLD).then_some((cluster, similarity)) + }); + match joined { + Some((cluster, similarity)) => cluster.push((record, similarity)), + None => clusters.push(vec![(record, 1.0)]), + } + } + + for state in ["pending", "accepted", "rejected"] { + let directory = candidates_root.join(state); + fs::create_dir_all(&directory)?; + harden_directory(&directory)?; + } + harden_directory(candidates_root)?; + + let mut written = 0; + for cluster in &clusters { + if let Some(candidate) = candidate_for(cluster) { + if write_candidate(candidates_root, candidate)? { + written += 1; + } + } + } + Ok(written) +} + +fn candidate_for(cluster: &[(&LessonRecord, f64)]) -> Option { + let anchor = cluster[0].0; + let projects: BTreeSet<&str> = cluster.iter().map(|(r, _)| r.project_id.as_str()).collect(); + let has_tag = |wanted: &[&str]| { + cluster.iter().any(|(r, _)| { + r.tags + .iter() + .any(|tag| wanted.iter().any(|w| tag.eq_ignore_ascii_case(w))) + }) + }; + let tagged_global = has_tag(GLOBAL_TAGS); + let tagged_user = has_tag(USER_TAGS); + + let mut reasons = Vec::new(); + if projects.len() >= 2 { + reasons.push("recurrence".to_owned()); + } + if tagged_global { + reasons.push("tagged-global".to_owned()); + } + if names_dependency_or_cli(&anchor.text) + && !cluster.iter().any(|(r, _)| has_repo_relative_path(&r.text)) + { + reasons.push("portable-tooling".to_owned()); + } + if reasons.is_empty() { + return None; + } + + let scope_tags: Vec<&str> = GLOBAL_TAGS.iter().chain(USER_TAGS).copied().collect(); + let tags: BTreeSet = cluster + .iter() + .flat_map(|(r, _)| r.tags.iter()) + .filter(|tag| !scope_tags.iter().any(|s| tag.eq_ignore_ascii_case(s))) + .cloned() + .collect(); + let now = Utc::now().to_rfc3339(); + Some(Candidate { + schema_version: 1, + id: format!("promo-{}", &hex_digest(&anchor.lesson_id)[..16]), + kind: "promotion".to_owned(), + state: "pending".to_owned(), + scope: if tagged_user && !tagged_global { + "user" + } else { + "global" + } + .to_owned(), + reasons, + title: title_for(&anchor.text), + content: anchor.text.clone(), + tags: tags.into_iter().collect(), + fingerprint: anchor.lesson_id.clone(), + evidence: cluster + .iter() + .map(|(r, similarity)| Evidence { + lesson_id: r.lesson_id.clone(), + project_id: r.project_id.clone(), + project_root: r.project_root.clone(), + source: r.source.clone(), + session_id: r.session_id.clone(), + recorded_at: r.recorded_at.clone(), + similarity: (similarity * 1000.0).round() / 1000.0, + }) + .collect(), + created_at: now.clone(), + updated_at: now, + }) +} + +/// Write a candidate unless it was already decided or is unchanged. Returns +/// whether a file was written. +fn write_candidate(root: &Path, mut candidate: Candidate) -> Result { + let filename = format!("{}.json", candidate.id); + if root.join("accepted").join(&filename).exists() + || root.join("rejected").join(&filename).exists() + { + return Ok(false); + } + let pending = root.join("pending").join(&filename); + if pending.exists() { + let existing: Value = serde_json::from_slice(&fs::read(&pending)?)?; + let existing_ids: BTreeSet<&str> = existing["evidence"] + .as_array() + .map(|items| { + items + .iter() + .filter_map(|item| item["lessonId"].as_str()) + .collect() + }) + .unwrap_or_default(); + let new_ids: BTreeSet<&str> = candidate + .evidence + .iter() + .map(|e| e.lesson_id.as_str()) + .collect(); + if existing_ids == new_ids { + return Ok(false); + } + if let Some(created_at) = existing["createdAt"].as_str() { + candidate.created_at = created_at.to_owned(); + } + } + atomic_json(&pending, &candidate)?; + Ok(true) +} + +fn extract_lessons(message: &str) -> Vec { + let mut lessons = Vec::new(); + for line in message.lines() { + let line = line.trim().trim_start_matches(['-', '*', ' ']).trim(); + for prefix in LESSON_PREFIXES { + let matches = line + .get(..prefix.len()) + .is_some_and(|head| head.eq_ignore_ascii_case(prefix)); + if matches { + let body = line[prefix.len()..].trim(); + if !body.is_empty() { + lessons.push(body.chars().take(LESSON_TEXT_LIMIT).collect()); + } + break; + } + } + } + if lessons.is_empty() { + let whole = message.trim(); + if !whole.is_empty() { + lessons.push(whole.chars().take(LESSON_TEXT_LIMIT).collect()); + } + } + lessons +} + +/// Inline `#tag` words, e.g. `#global` or `#scope:user`. +fn hashtags(text: &str) -> Vec { + let mut tags: Vec = text + .split_whitespace() + .filter_map(|word| word.strip_prefix('#')) + .map(|tag| tag.trim_end_matches(|c: char| !(c.is_alphanumeric() || c == ':'))) + .filter(|tag| { + !tag.is_empty() + && tag + .chars() + .all(|c| c.is_ascii_alphanumeric() || c == ':' || c == '-') + }) + .map(str::to_ascii_lowercase) + .collect(); + tags.sort(); + tags.dedup(); + tags +} + +fn tokens(text: &str) -> Vec { + text.split(|c: char| !c.is_alphanumeric()) + .filter(|token| !token.is_empty()) + .map(str::to_lowercase) + .collect() +} + +fn shingles(tokens: &[String]) -> Vec { + let mut hashes: Vec = if tokens.len() < SHINGLE_SIZE { + vec![shingle_hash(&tokens.join(" "))] + } else { + tokens + .windows(SHINGLE_SIZE) + .map(|window| shingle_hash(&window.join(" "))) + .collect() + }; + hashes.sort_unstable(); + hashes.dedup(); + hashes +} + +/// First 8 bytes of SHA-256: stable across Rust releases, unlike `DefaultHasher`. +fn shingle_hash(shingle: &str) -> u64 { + let digest = Sha256::digest(shingle.as_bytes()); + let mut bytes = [0_u8; 8]; + bytes.copy_from_slice(&digest[..8]); + u64::from_be_bytes(bytes) +} + +/// Jaccard similarity of two sorted, de-duplicated sets. +fn jaccard(left: &[u64], right: &[u64]) -> f64 { + if left.is_empty() && right.is_empty() { + return 0.0; + } + let (mut i, mut j, mut shared) = (0, 0, 0usize); + while i < left.len() && j < right.len() { + match left[i].cmp(&right[j]) { + std::cmp::Ordering::Less => i += 1, + std::cmp::Ordering::Greater => j += 1, + std::cmp::Ordering::Equal => { + shared += 1; + i += 1; + j += 1; + } + } + } + shared as f64 / (left.len() + right.len() - shared) as f64 +} + +fn names_dependency_or_cli(text: &str) -> bool { + let in_code_span = text.split('`').skip(1).step_by(2).any(|span| { + let span = span.trim(); + let first = span.split_whitespace().next().unwrap_or(""); + KNOWN_CLIS.contains(&first.to_ascii_lowercase().as_str()) + || is_versioned_name(first) + || is_manifest_pin(span) + }); + in_code_span + || text + .split_whitespace() + .map(|word| word.trim_matches(|c: char| "`'\"(),;".contains(c))) + .any(is_versioned_name) +} + +/// `name@1.2.3`, including scoped `@scope/name@1.2`. +fn is_versioned_name(word: &str) -> bool { + let Some((name, version)) = word.rsplit_once('@') else { + return false; + }; + let name = name.trim_start_matches('@'); + !name.is_empty() + && name + .chars() + .all(|c| c.is_ascii_alphanumeric() || "-_./".contains(c)) + && version + .trim_start_matches(['^', '~', '=', 'v']) + .starts_with(|c: char| c.is_ascii_digit()) +} + +/// A manifest pin such as `tokio = "1.40"` or `flutter_rust_bridge = "=2.12.0"`. +fn is_manifest_pin(span: &str) -> bool { + let Some((name, value)) = span.split_once('=') else { + return false; + }; + let name = name.trim(); + !name.is_empty() + && name + .chars() + .all(|c| c.is_ascii_alphanumeric() || "-_".contains(c)) + && value + .trim() + .trim_start_matches('"') + .trim_start_matches(['^', '~', '=', '<', '>']) + .starts_with(|c: char| c.is_ascii_digit()) +} + +fn has_repo_relative_path(text: &str) -> bool { + text.split_whitespace().any(|word| { + let word = word.trim_matches(|c: char| "`'\"()[]{},;".contains(c)); + let word = word.split(':').next().unwrap_or("").trim_end_matches('.'); + if !word.contains('/') || word.contains("://") || word.starts_with(['/', '~', '$', '@']) { + return false; + } + if word.starts_with("./") || word.starts_with("../") { + return true; + } + let segments: Vec<&str> = word.split('/').filter(|s| !s.is_empty()).collect(); + if segments.len() < 2 { + return false; + } + let last = segments[segments.len() - 1]; + REPO_DIRS.contains(&segments[0]) + || last + .rsplit_once('.') + .is_some_and(|(stem, ext)| !stem.is_empty() && !ext.is_empty()) + }) +} + +fn title_for(text: &str) -> String { + let first_line = text.lines().next().unwrap_or("").trim(); + let mut title: String = first_line.chars().take(80).collect(); + if first_line.chars().count() > 80 { + title.push('…'); + } + if title.is_empty() { + "Promoted lesson".to_owned() + } else { + title + } +} + +fn hex_digest(value: &str) -> String { + Sha256::digest(value.as_bytes()) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect() +} diff --git a/pk-learning-worker/tests/promotion.rs b/pk-learning-worker/tests/promotion.rs new file mode 100644 index 0000000..ddbfb06 --- /dev/null +++ b/pk-learning-worker/tests/promotion.rs @@ -0,0 +1,141 @@ +//! The promotion detector: a lesson that recurs in two projects becomes exactly +//! one pending promotion candidate carrying one evidence entry per project, and +//! re-running the worker proposes nothing new. + +use serde_json::{json, Value}; +use std::{ + fs, + path::{Path, PathBuf}, + process::Command, +}; + +const LESSON: &str = + "LESSON: anchor hook matchers to the plugin prefix so renamed agents still match their hooks"; + +fn transcript(dir: &Path, name: &str, text: &str) -> PathBuf { + let path = dir.join(format!("transcript-{name}.jsonl")); + fs::write( + &path, + serde_json::to_string(&json!({ + "type": "assistant", + "message": {"role": "assistant", "content": [{"type": "text", "text": text}]} + })) + .unwrap() + + "\n", + ) + .unwrap(); + path +} + +fn queue_job(queue: &Path, event_id: &str, project: &Path, transcript: &Path) { + let job = json!({ + "schemaVersion": 2, + "eventId": event_id, + "eventType": "subagent_stop", + "harness": "fixture", + "sessionId": format!("session-{}", &event_id[..8]), + "projectRoot": project, + "transcriptPath": transcript, + "capturedAt": "2026-10-04T00:00:00Z", + "payloadDigest": "fixture", + "attempt": 0 + }); + fs::write( + queue.join("pending").join(format!("{event_id}.json")), + serde_json::to_vec_pretty(&job).unwrap(), + ) + .unwrap(); +} + +fn run_worker(home: &Path) { + let output = Command::new(env!("CARGO_BIN_EXE_prometheus-learning-worker")) + .env("HOME", home) + .env("RUST_LOG", "error") + .env_remove("PROMETHEUS_LEARNING_QUEUE") + .args(["--memory-url", "http://127.0.0.1:1", "run-once"]) + .output() + .unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); +} + +fn json_files(directory: &Path) -> Vec { + let mut files: Vec = fs::read_dir(directory) + .unwrap() + .flatten() + .map(|entry| entry.path()) + .filter(|path| path.extension().and_then(|value| value.to_str()) == Some("json")) + .collect(); + files.sort(); + files +} + +#[test] +fn a_lesson_recurring_in_two_projects_yields_one_candidate_and_reruns_add_none() { + let fixture = tempfile::tempdir().unwrap(); + let home = fixture.path().join("home"); + let queue = home.join(".prometheus/learning-queue"); + fs::create_dir_all(queue.join("pending")).unwrap(); + let mut projects = Vec::new(); + for name in ["alpha", "beta"] { + let project = fixture.path().join(name); + fs::create_dir_all(project.join(".git")).unwrap(); + projects.push(project); + } + queue_job( + &queue, + &"a".repeat(64), + &projects[0], + &transcript(fixture.path(), "alpha", LESSON), + ); + queue_job( + &queue, + &"b".repeat(64), + &projects[1], + &transcript(fixture.path(), "beta", LESSON), + ); + + run_worker(&home); + + let pending_dir = home.join(".prometheus/promotion-candidates/pending"); + let pending = json_files(&pending_dir); + assert_eq!(pending.len(), 1, "expected one candidate: {pending:?}"); + let candidate: Value = serde_json::from_slice(&fs::read(&pending[0]).unwrap()).unwrap(); + assert_eq!(candidate["kind"], "promotion", "{candidate:#}"); + assert_eq!(candidate["state"], "pending", "{candidate:#}"); + let reasons = candidate["reasons"].as_array().unwrap(); + assert!( + reasons.iter().any(|reason| reason == "recurrence"), + "{candidate:#}" + ); + let evidence = candidate["evidence"].as_array().unwrap(); + assert_eq!(evidence.len(), 2, "{candidate:#}"); + let projects_cited: std::collections::BTreeSet<&str> = evidence + .iter() + .map(|item| item["projectId"].as_str().unwrap()) + .collect(); + assert_eq!(projects_cited.len(), 2, "{candidate:#}"); + assert_eq!( + pending[0].file_stem().unwrap().to_str().unwrap(), + candidate["id"].as_str().unwrap() + ); + + let index = home.join(".prometheus/learning-index/lessons.jsonl"); + let indexed = fs::read_to_string(&index).unwrap(); + assert_eq!(indexed.lines().count(), 2, "{indexed}"); + let before = fs::read(&pending[0]).unwrap(); + + run_worker(&home); + + let after_pending = json_files(&pending_dir); + assert_eq!(after_pending, pending, "a re-run must not add candidates"); + assert_eq!( + fs::read(&pending[0]).unwrap(), + before, + "an unchanged candidate is not rewritten" + ); + assert_eq!(fs::read_to_string(&index).unwrap(), indexed); +}