diff --git a/Cargo.lock b/Cargo.lock index 3c292f1..184857a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -358,6 +358,7 @@ dependencies = [ "chrono", "coven-github-api", "rusqlite", + "serde", "serde_json", "tokio", "tracing", diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index 2e58291..4bdaf18 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -11,7 +11,7 @@ use tower_http::trace::TraceLayer; use tracing_subscriber::EnvFilter; use coven_github_config::Config; -use coven_github_webhook::routes::{handle_webhook, list_tasks, AppState}; +use coven_github_webhook::routes::{handle_webhook, list_memory, list_tasks, AppState}; use coven_github_worker as worker; #[derive(Parser)] @@ -119,6 +119,7 @@ async fn main() -> Result<()> { let app = Router::new() .route("/webhook", post(handle_webhook)) .route("/api/github/tasks", get(list_tasks)) + .route("/api/github/memory", get(list_memory)) .with_state(state) .layer(TraceLayer::new_for_http()); diff --git a/crates/store/Cargo.toml b/crates/store/Cargo.toml index 19d505c..86e84c0 100644 --- a/crates/store/Cargo.toml +++ b/crates/store/Cargo.toml @@ -8,6 +8,7 @@ license.workspace = true anyhow.workspace = true chrono.workspace = true rusqlite.workspace = true +serde.workspace = true serde_json.workspace = true tokio.workspace = true tracing.workspace = true diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index 85eb8f4..cf0f09c 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -15,7 +15,7 @@ use std::path::Path; use std::sync::{Arc, Mutex}; /// Current schema version, stored in `PRAGMA user_version`. -const SCHEMA_VERSION: i32 = 3; +const SCHEMA_VERSION: i32 = 4; /// Handle to the durable store. Cheap to clone; all clones share one writer /// connection. @@ -483,6 +483,101 @@ impl Store { .await .expect("store task panicked") } + + /// Records memory activity for the inspect/audit trail (issue #6). Every + /// read and write the runtime reported is recorded with the adapter's + /// verdict, so a customer can later see what a familiar read from and + /// attempted to write to their repository's memory. + pub async fn record_memory_activity(&self, rows: Vec) -> Result<()> { + if rows.is_empty() { + return Ok(()); + } + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || { + let mut conn = conn.lock().expect("store mutex poisoned"); + let tx = conn.transaction()?; + for row in &rows { + tx.execute( + "INSERT INTO memory_activity + (at, installation_id, repo, task_id, op, target, scope, outcome) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", + params![ + now_rfc3339(), + row.installation_id, + row.repo, + row.task_id, + row.op, + row.target, + row.scope, + row.outcome, + ], + )?; + } + tx.commit()?; + Ok(()) + }) + .await + .expect("store task panicked") + } + + /// Lists recorded memory activity, newest first, narrowed to the caller's + /// [`ApiScope`] (issue #6). `None` scope (service/open callers) sees all. + pub async fn list_memory(&self, scope: Option) -> Result> { + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || { + let conn = conn.lock().expect("store mutex poisoned"); + let installation_filter = scope.as_ref().map(|s| s.installation_id); + let mut stmt = conn.prepare( + "SELECT at, installation_id, repo, task_id, op, target, scope, outcome + FROM memory_activity + WHERE (?1 IS NULL OR installation_id = ?1) + ORDER BY id DESC", + )?; + let rows = stmt + .query_map(params![installation_filter], |row| { + Ok(MemoryActivity { + at: row.get(0)?, + installation_id: row.get(1)?, + repo: row.get(2)?, + task_id: row.get(3)?, + op: row.get(4)?, + target: row.get(5)?, + scope: row.get(6)?, + outcome: row.get(7)?, + }) + })? + .collect::, _>>()?; + // Repository narrowing within the installation, when scoped. + let filtered = rows + .into_iter() + .filter(|r| match scope.as_ref().and_then(|s| s.repos.as_ref()) { + Some(repos) => repos.contains(&r.repo), + None => true, + }) + .collect(); + Ok(filtered) + }) + .await + .expect("store task panicked") + } +} + +/// One recorded memory operation for the inspect/audit trail (issue #6). +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)] +pub struct MemoryActivity { + /// RFC 3339 timestamp; set by the store on insert (ignored on the way in). + pub at: String, + pub installation_id: u64, + pub repo: String, + pub task_id: String, + /// `read` or `write`. + pub op: String, + /// The memory entry id/key. + pub target: String, + /// Namespace: `repo` | `tenant_shared` | `familiar_global`. + pub scope: String, + /// Adapter verdict: `accepted` or `rejected:`. + pub outcome: String, } /// Terminal transition applied by [`Store::finish`]. @@ -613,6 +708,29 @@ fn migrate(conn: &Connection) -> Result<()> { ) .context("failed to apply schema v3")?; } + if version < 4 { + // v4: memory governance audit for inspect (issue #6). Records every + // memory read/write the runtime reported, with the adapter's verdict, + // scoped to installation/repo so customers can inspect it. + conn.execute_batch( + r#" + CREATE TABLE memory_activity ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + at TEXT NOT NULL, + installation_id INTEGER NOT NULL, + repo TEXT NOT NULL, + task_id TEXT NOT NULL, + op TEXT NOT NULL, + target TEXT NOT NULL, + scope TEXT NOT NULL, + outcome TEXT NOT NULL + ); + CREATE INDEX memory_activity_scope + ON memory_activity(installation_id, repo); + "#, + ) + .context("failed to apply schema v4")?; + } conn.pragma_update(None, "user_version", SCHEMA_VERSION)?; Ok(()) } @@ -734,6 +852,72 @@ mod tests { } } + fn memory_row(installation_id: u64, repo: &str, target: &str, outcome: &str) -> MemoryActivity { + MemoryActivity { + at: String::new(), + installation_id, + repo: repo.to_string(), + task_id: "task-1".to_string(), + op: "write".to_string(), + target: target.to_string(), + scope: "repo".to_string(), + outcome: outcome.to_string(), + } + } + + #[tokio::test] + async fn memory_activity_records_and_lists_newest_first_with_verdicts() { + let store = Store::open_in_memory().unwrap(); + store + .record_memory_activity(vec![ + memory_row(1, "OpenCoven/demo", "repo/OpenCoven/demo/a", "accepted"), + memory_row(1, "OpenCoven/demo", "repo/other/x", "rejected:out_of_scope"), + ]) + .await + .unwrap(); + + let all = store.list_memory(None).await.unwrap(); + assert_eq!(all.len(), 2); + // Newest first: the rejected row was inserted second. + assert_eq!(all[0].outcome, "rejected:out_of_scope"); + assert!(!all[0].at.is_empty(), "the store stamps the timestamp"); + } + + #[tokio::test] + async fn list_memory_is_scoped_to_installation_and_repo() { + let store = Store::open_in_memory().unwrap(); + store + .record_memory_activity(vec![ + memory_row(1, "OpenCoven/billing", "repo/OpenCoven/billing/a", "accepted"), + memory_row(1, "OpenCoven/web", "repo/OpenCoven/web/b", "accepted"), + memory_row(2, "OpenCoven/other", "repo/OpenCoven/other/c", "accepted"), + ]) + .await + .unwrap(); + + // Tenant scoped to installation 1 never sees installation 2. + let inst1 = store + .list_memory(Some(ApiScope { + installation_id: 1, + repos: None, + })) + .await + .unwrap(); + assert_eq!(inst1.len(), 2); + assert!(inst1.iter().all(|r| r.installation_id == 1)); + + // Narrowed further to a single repo. + let one_repo = store + .list_memory(Some(ApiScope { + installation_id: 1, + repos: Some(vec!["OpenCoven/billing".to_string()]), + })) + .await + .unwrap(); + assert_eq!(one_repo.len(), 1); + assert_eq!(one_repo[0].repo, "OpenCoven/billing"); + } + fn review_task(id: &str, pr: u64) -> Task { Task { kind: TaskKind::ReviewPullRequest { diff --git a/crates/webhook/src/routes.rs b/crates/webhook/src/routes.rs index a6434bc..8730bb4 100644 --- a/crates/webhook/src/routes.rs +++ b/crates/webhook/src/routes.rs @@ -115,6 +115,72 @@ pub async fn list_tasks(State(state): State, headers: HeaderMap) -> im } } +/// GET /api/github/memory — memory activity for the tenant boundary (issue #6), +/// so a customer can inspect what memory a familiar read from and attempted to +/// write to their repositories. Same auth as `list_tasks`: `token` mode fails +/// closed, a tenant sees only its own installation, and every read is audited. +pub async fn list_memory(State(state): State, headers: HeaderMap) -> impl IntoResponse { + let action = "list_memory"; + let (caller, scope) = match authorize_api(&state.config.api, &headers) { + ApiCaller::Denied => { + if let Err(e) = state + .store + .record_api_read("anonymous", "none", action, "denied") + .await + { + error!("api audit write failed: {e:#}"); + } + return ( + StatusCode::UNAUTHORIZED, + Json(json!({"ok": false, "error": "unauthorized"})), + ) + .into_response(); + } + ApiCaller::Open => ("open".to_string(), None), + ApiCaller::Service => ("service".to_string(), None), + ApiCaller::Tenant(tenant) => ( + format!("tenant:{}", tenant.installation_id), + Some(ApiScope { + installation_id: tenant.installation_id, + repos: if tenant.repos.is_empty() { + None + } else { + Some(tenant.repos.clone()) + }, + }), + ), + }; + let scope_label = scope + .as_ref() + .map(|s| format!("installation:{}", s.installation_id)) + .unwrap_or_else(|| "all".to_string()); + + match state.store.list_memory(scope).await { + Ok(entries) => { + if let Err(e) = state + .store + .record_api_read(&caller, &scope_label, action, &format!("ok:{}", entries.len())) + .await + { + error!("api audit write failed: {e:#}"); + } + Json(json!({ "ok": true, "memory": entries })).into_response() + } + Err(e) => { + error!("memory list unavailable: {e:#}"); + let _ = state + .store + .record_api_read(&caller, &scope_label, action, "error") + .await; + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"ok": false, "error": "memory list unavailable"})), + ) + .into_response() + } + } +} + /// Resolved identity of a task-API caller. enum ApiCaller<'a> { /// `open` mode: unauthenticated local development. @@ -1590,6 +1656,70 @@ mod tenancy_tests { ); } + async fn list_mem(state: &AppState, bearer: Option<&str>) -> (StatusCode, serde_json::Value) { + let mut headers = HeaderMap::new(); + if let Some(token) = bearer { + headers.insert( + "authorization", + format!("Bearer {token}").parse().expect("header"), + ); + } + let response = list_memory(State(state.clone()), headers) + .await + .into_response(); + let status = response.status(); + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .expect("body"); + (status, serde_json::from_slice(&bytes).expect("json")) + } + + async fn seed_memory(state: &AppState, installation_id: u64, repo: &str) { + state + .store + .record_memory_activity(vec![coven_github_store::MemoryActivity { + at: String::new(), + installation_id, + repo: repo.to_string(), + task_id: "t".to_string(), + op: "read".to_string(), + target: format!("repo/{repo}/x"), + scope: "repo".to_string(), + outcome: "accepted".to_string(), + }]) + .await + .expect("record memory"); + } + + #[tokio::test] + async fn memory_inspect_is_tenant_scoped_and_fails_closed() { + let state = token_state(two_tenant_api()); + seed_memory(&state, 1, "OpenCoven/alpha").await; + seed_memory(&state, 2, "OpenCoven/beta").await; + + // No / wrong token in token mode → fail closed, no data leaks. + for bearer in [None, Some("wrong-token-0123456789abcdef")] { + let (status, json) = list_mem(&state, bearer).await; + assert_eq!(status, StatusCode::UNAUTHORIZED, "bearer: {bearer:?}"); + assert!(json.get("memory").is_none(), "no data may leak: {json}"); + } + + // Installation 1's token sees only its own memory activity. + let (status, json) = list_mem(&state, Some("tenant-one-0123456789abcdef")).await; + assert_eq!(status, StatusCode::OK); + let repos: Vec<&str> = json["memory"] + .as_array() + .unwrap() + .iter() + .map(|m| m["repo"].as_str().unwrap()) + .collect(); + assert_eq!(repos, vec!["OpenCoven/alpha"]); + + // The service token sees both installations. + let (_, json) = list_mem(&state, Some("service-token-0123456789abcdef")).await; + assert_eq!(json["memory"].as_array().unwrap().len(), 2); + } + #[tokio::test] async fn tenant_repo_scope_narrows_within_the_installation() { let mut api = two_tenant_api(); diff --git a/crates/worker/src/lib.rs b/crates/worker/src/lib.rs index 6b53401..11696a4 100644 --- a/crates/worker/src/lib.rs +++ b/crates/worker/src/lib.rs @@ -338,6 +338,7 @@ async fn execute_task_with_minter( &targets, &workspace, check_id, + &store, ) .await; @@ -642,6 +643,56 @@ impl Minter { /// and retry-safe session failures that exhausted the retry budget. Cosmetic /// side-effects (comments, the in-progress transition, even PR opening) are /// best-effort and logged rather than propagated. +#[allow(clippy::too_many_arguments)] +/// Maps the runtime's reported memory activity to audit rows, stamping each +/// with the adapter's verdict (`accepted` or `rejected:`) from the +/// validation pass (issue #6). +fn memory_activity_rows( + installation_id: u64, + repo: &str, + task_id: &str, + used: &coven_github_api::MemoryUsed, + rejections: &[memory::MemoryRejection], +) -> Vec { + let verdict = |op: memory::MemoryOp, target: &str| { + rejections + .iter() + .find(|r| r.op == op && r.target == target) + .map(|r| format!("rejected:{}", r.reason)) + .unwrap_or_else(|| "accepted".to_string()) + }; + let row = |op: &str, target: &str, scope: &str, outcome: String| { + coven_github_store::MemoryActivity { + at: String::new(), + installation_id, + repo: repo.to_string(), + task_id: task_id.to_string(), + op: op.to_string(), + target: target.to_string(), + scope: scope.to_string(), + outcome, + } + }; + let mut rows = Vec::new(); + for entry in &used.read { + rows.push(row( + "read", + &entry.id, + &entry.scope, + verdict(memory::MemoryOp::Read, &entry.id), + )); + } + for write in &used.proposed { + rows.push(row( + "write", + &write.key, + &write.scope, + verdict(memory::MemoryOp::Write, &write.key), + )); + } + rows +} + #[allow(clippy::too_many_arguments)] async fn run_and_publish( config: &Config, @@ -653,6 +704,7 @@ async fn run_and_publish( targets: &ResolvedTargets, workspace: &Path, check_id: u64, + store: &Store, ) -> Result { // Provision ephemeral workspace and write the tokenless session brief. tokio::fs::create_dir_all(workspace).await?; @@ -796,6 +848,13 @@ async fn run_and_publish( "refused out-of-policy memory activity — not persisting those entries" ); } + // Record every reported read/write with the adapter's verdict so a + // customer can inspect what memory a familiar used on their repo (#6). + let activity = + memory_activity_rows(task.installation_id, &repo_full, &task.id, used, &rejections); + if let Err(e) = store.record_memory_activity(activity).await { + warn!(task_id = %task.id, "failed to record memory activity: {e:#}"); + } } // Stale-ref gate (issue #8): review findings are only valid for the head @@ -2197,6 +2256,7 @@ exit 0 head_is_fork: false, }; let workspace = root.join("ws"); + let store = coven_github_store::Store::open_in_memory().expect("store"); let published = run_and_publish( &config, @@ -2208,6 +2268,7 @@ exit 0 &targets, &workspace, 7, + &store, ) .await .expect("publication should succeed"); diff --git a/crates/worker/src/memory.rs b/crates/worker/src/memory.rs index 6a94a7f..066980b 100644 --- a/crates/worker/src/memory.rs +++ b/crates/worker/src/memory.rs @@ -290,6 +290,36 @@ mod tests { assert!(compute_policy(inputs(TrustScope::Maintainer, false)).is_none()); } + #[test] + fn activity_rows_carry_the_adapter_verdict() { + use crate::memory_activity_rows; + let policy = compute_policy(inputs(TrustScope::ForkPr, true)).unwrap(); + let used = used_with( + // A fork write is rejected (no write scope); the read is accepted. + vec![proposed("repo", "repo/acme/billing/y", "pending")], + vec![MemoryEntryRef { + id: "repo/acme/billing/x".to_string(), + scope: "repo".to_string(), + }], + ); + let rejections = validate_memory_used(&policy, &used, never_secret); + let rows = memory_activity_rows(42, "acme/billing", "task-1", &used, &rejections); + + assert_eq!(rows.len(), 2); + assert_eq!( + rows.iter().find(|r| r.op == "read").unwrap().outcome, + "accepted" + ); + assert!( + rows.iter() + .find(|r| r.op == "write") + .unwrap() + .outcome + .starts_with("rejected:"), + "fork write should be recorded as rejected" + ); + } + #[test] fn fork_review_overrides_actor_trust() { // A fork PR is untrusted content: even a maintainer command reviewing it