Skip to content

Add memory tools and ingestion #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rmzlb wants to merge 1 commit into
base: baaton/pr2-memory-layer
Choose a base branch
from
Open

Add memory tools and ingestion #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
111 changes: 78 additions & 33 deletions backend/src/github/webhook_processor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,18 +16,16 @@ async fn record_github_activity(
action: &str,
metadata: serde_json::Value,
) {
let org_id: Option<String> = sqlx::query_scalar(
"SELECT org_id FROM projects WHERE id = $1"
)
.bind(project_id)
.fetch_optional(pool)
.await
.ok()
.flatten();
let org_id: Option<String> = sqlx::query_scalar("SELECT org_id FROM projects WHERE id = $1")
.bind(project_id)
.fetch_optional(pool)
.await
.ok()
.flatten();

let Some(org_id) = org_id else { return };

let user_id = format!("github:{}", sender_login);
let user_id = format!("github:{}", sender_login);
let user_name = format!("@{}", sender_login);

crate::routes::activity::log_activity(
Expand All @@ -38,26 +36,24 @@ async fn record_github_activity(
&user_id,
Some(&user_name),
action,
None, None, None,
None,
None,
None,
Some(metadata),
).await;
)
.await;
}

/// Process a webhook event that was previously stored in github_webhook_events.
///
/// Called from the webhook handler's spawned task AND from the job runner
/// (for retries of failed events).
pub async fn process_webhook_event(
pool: &PgPool,
delivery_id: &str,
) -> Result<(), anyhow::Error> {
pub async fn process_webhook_event(pool: &PgPool, delivery_id: &str) -> Result<(), anyhow::Error> {
// Mark as processing
sqlx::query(
"UPDATE github_webhook_events SET status = 'processing' WHERE delivery_id = $1",
)
.bind(delivery_id)
.execute(pool)
.await?;
sqlx::query("UPDATE github_webhook_events SET status = 'processing' WHERE delivery_id = $1")
.bind(delivery_id)
.execute(pool)
.await?;

// Fetch the event
let event = sqlx::query_as::<_, GitHubWebhookEvent>(
Expand Down Expand Up @@ -91,7 +87,11 @@ pub async fn process_webhook_event(
}
Err(ref e) => {
let retry_count = event.retry_count + 1;
let new_status = if retry_count >= 3 { "failed" } else { "pending" };
let new_status = if retry_count >= 3 {
"failed"
} else {
"pending"
};

sqlx::query(
r#"UPDATE github_webhook_events
Expand Down Expand Up @@ -359,7 +359,7 @@ async fn handle_pull_request_event(

// Record PR activity (opened or merged only — not every draft update)
let gh_action = match (action, pr["merged"].as_bool()) {
("opened", _) => Some("github_pr_opened"),
("opened", _) => Some("github_pr_opened"),
("closed", Some(true)) => Some("github_pr_merged"),
_ => None,
};
Expand All @@ -377,7 +377,55 @@ async fn handle_pull_request_event(
"pr_title": pr_title,
"branch": head_branch,
}),
).await;
)
.await;

let memory_org_id: Option<String> =
sqlx::query_scalar("SELECT org_id FROM projects WHERE id = $1")
.bind(mapping.project_id)
.fetch_optional(pool)
.await
.ok()
.flatten();
if let Some(memory_org_id) = memory_org_id {
crate::routes::memory::record_memory_best_effort(
pool,
crate::routes::memory::NewMemory {
org_id: memory_org_id,
project_id: Some(mapping.project_id),
source: "github".to_string(),
kind: if gh_action == "github_pr_merged" {
"learning".to_string()
} else {
"note".to_string()
},
content: format!(
"GitHub PR #{} {} for linked issue: {} ({})",
pr_number,
if gh_action == "github_pr_merged" {
"merged"
} else {
"opened"
},
pr_title,
head_branch
),
tags: vec!["github".to_string(), "pull-request".to_string()],
confidence: 0.85,
external_url: pr["html_url"].as_str().map(|s| s.to_string()),
metadata: serde_json::json!({
"issue_id": issue_id,
"repo": pr["base"]["repo"]["full_name"].as_str().unwrap_or(""),
"pr_number": pr_number,
"action": gh_action,
"branch": head_branch,
}),
created_by: Some(format!("github:{}", sender_login)),
created_by_name: Some(format!("@{}", sender_login)),
},
)
.await;
}
}

Ok(())
Expand Down Expand Up @@ -420,10 +468,7 @@ async fn handle_pr_review_event(

// ─── Push Events (Commits) ────────────────────────────

async fn handle_push_event(
pool: &PgPool,
event: &GitHubWebhookEvent,
) -> Result<(), anyhow::Error> {
async fn handle_push_event(pool: &PgPool, event: &GitHubWebhookEvent) -> Result<(), anyhow::Error> {
let payload = &event.payload;
let repo = &payload["repository"];
let github_repo_id = repo["id"].as_i64().unwrap_or(0);
Expand Down Expand Up @@ -456,10 +501,9 @@ async fn handle_push_event(
let timestamp = commit["timestamp"].as_str().unwrap_or("");

// Try to find linked issue from branch name or commit message
let issue_id = crate::github::issue_linker::find_linked_issue(
pool, &mapping, branch, message, "",
)
.await?;
let issue_id =
crate::github::issue_linker::find_linked_issue(pool, &mapping, branch, message, "")
.await?;

let issue_id = match issue_id {
Some(id) => id,
Expand Down Expand Up @@ -498,7 +542,8 @@ async fn handle_push_event(
"sha": sha,
"message": message,
}),
).await;
)
.await;
}

Ok(())
Expand Down
121 changes: 119 additions & 2 deletions backend/src/routes/ai_agent.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,12 +13,21 @@ use axum::{
};
use futures::stream::Stream;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use serde_json::{json, Value};
use sqlx::PgPool;
use std::{convert::Infallible, time::Duration};

use crate::middleware::AuthUser;

fn truncate(s: &str, max: usize) -> String {
if s.chars().count() <= max {
s.to_string()
} else {
let truncated: String = s.chars().take(max.saturating_sub(1)).collect();
format!("{}…", truncated)
}
}

// ─── Request / Response Types ─────────────────────────────────────────────────

#[derive(Debug, Deserialize)]
Expand Down Expand Up @@ -276,6 +285,28 @@ async fn build_project_context(pool: &PgPool, org_id: &str, project_ids: &[Strin
cnt: i64,
}

#[derive(sqlx::FromRow)]
struct ContextRow {
project_id: uuid::Uuid,
stack: Option<String>,
conventions: Option<String>,
architecture: Option<String>,
constraints: Option<String>,
current_focus: Option<String>,
learnings: Option<String>,
}

#[derive(sqlx::FromRow)]
struct MemoryRow {
project_id: uuid::Uuid,
source: String,
kind: String,
content: String,
tags: Vec<String>,
confidence: f64,
created_at: chrono::DateTime<chrono::Utc>,
}

let projects: Vec<ProjectRow> = if project_ids.is_empty() {
sqlx::query_as::<_, ProjectRow>(
"SELECT id, name, prefix FROM projects WHERE org_id = $1 ORDER BY name ASC LIMIT 20",
Expand Down Expand Up @@ -333,10 +364,46 @@ async fn build_project_context(pool: &PgPool, org_id: &str, project_ids: &[Strin
.unwrap_or_default()
};

let project_uuid_list: Vec<uuid::Uuid> = projects.iter().map(|p| p.id).collect();

let project_contexts: Vec<ContextRow> = sqlx::query_as::<_, ContextRow>(
r#"
SELECT project_id, stack, conventions, architecture, constraints, current_focus, learnings
FROM project_contexts
WHERE org_id = $1 AND project_id = ANY($2::uuid[])
"#,
)
.bind(org_id)
.bind(&project_uuid_list)
.fetch_all(pool)
.await
.unwrap_or_default();

let memories: Vec<MemoryRow> = sqlx::query_as::<_, MemoryRow>(
r#"
SELECT project_id, source, kind, content, tags, confidence, created_at
FROM (
SELECT m.*, row_number() OVER (PARTITION BY project_id ORDER BY confidence DESC, created_at DESC) AS rn
FROM memories m
WHERE m.org_id = $1 AND m.project_id = ANY($2::uuid[])
) ranked
WHERE rn <= 5
ORDER BY project_id, confidence DESC, created_at DESC
"#,
)
.bind(org_id)
.bind(&project_uuid_list)
.fetch_all(pool)
.await
.unwrap_or_default();

// Build context string
let mut ctx = String::from("## Projets disponibles\n\n");
for project in &projects {
ctx.push_str(&format!("### {} (prefix: {}, id: {})\n", project.name, project.prefix, project.id));
ctx.push_str(&format!(
"### {} (prefix: {}, id: {})\n",
project.name, project.prefix, project.id
));
let project_counts: Vec<&IssueCountRow> = counts
.iter()
.filter(|c| c.project_name == project.name)
Expand All @@ -350,6 +417,56 @@ async fn build_project_context(pool: &PgPool, org_id: &str, project_ids: &[Strin
ctx.push_str(&format!(" - {}: {}\n", c.status, c.cnt));
}
}

if let Some(pc) = project_contexts.iter().find(|c| c.project_id == project.id) {
let fields = [
("Stack", pc.stack.as_deref()),
("Conventions", pc.conventions.as_deref()),
("Architecture", pc.architecture.as_deref()),
("Constraints", pc.constraints.as_deref()),
("Current focus", pc.current_focus.as_deref()),
("Learnings", pc.learnings.as_deref()),
];
let non_empty: Vec<(&str, &str)> = fields
.iter()
.filter_map(|(label, value)| {
(*value)
.map(str::trim)
.filter(|v| !v.is_empty())
.map(|v| (*label, v))
})
.collect();
if !non_empty.is_empty() {
ctx.push_str("- Project context:\n");
for (label, value) in non_empty {
ctx.push_str(&format!(" - {}: {}\n", label, truncate(value, 900)));
}
}
}

let project_memories: Vec<&MemoryRow> = memories
.iter()
.filter(|m| m.project_id == project.id)
.collect();
if !project_memories.is_empty() {
ctx.push_str("- Relevant memories:\n");
for m in project_memories {
let tags = if m.tags.is_empty() {
String::new()
} else {
format!(" tags={}", m.tags.join(","))
};
ctx.push_str(&format!(
" - [{} / {} / {:.2}{} / {}] {}\n",
m.kind,
m.source,
m.confidence,
tags,
m.created_at.format("%Y-%m-%d"),
truncate(&m.content, 500)
));
}
}
ctx.push('\n');
}

Expand Down
3 changes: 2 additions & 1 deletion backend/src/routes/ai_chat/stream.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,8 +42,9 @@ Lecture :
- **org_overview** = SANTÉ multi-projets (KPIs agrégés, action requise, sprints, milestones, rollup). 1ᵉʳ choix pour « état des lieux », « comment vont les projets », « dashboard santé ». 1 SEUL appel — jamais chaîné avec get_project_metrics+analyze_sprint.
- **weekly_recap** = ACTIVITÉ par période (qui a créé quoi, qui a changé quel statut, ce qui a landé). 1ᵉʳ choix pour « récap semaine », « qui a fait quoi », « tickets créés cette semaine », « changements de statut », « activité semaine ». Retourne les LISTES réelles avec auteurs.
- search_issues, get_project_metrics (drill-down 1 projet), analyze_sprint (1 sprint précis), suggest_priorities, find_similar_issues, workload_by_assignee, compare_projects (2-5 projets head-to-head), export_project.
- **search_memories** = mémoire projet durable (décisions, learnings, contraintes, risques, handoffs, intégrations). 1er choix pour « qu'est-ce qu'on sait déjà », « pourquoi », « gotcha », « remember ».

Écriture : enchaîner **propose_issue / propose_update_issue / propose_bulk_update / propose_comment** → validation UI → **create_issue, update_issue, bulk_update_issues, add_comment** avec les `finalValues` retournés. Planning : plan_milestones → create_milestones_batch, adjust_timeline. Autres : generate_prd, triage_issue, manage_* (initiatives, automations, SLA, templates, recurring).
Écriture : enchaîner **propose_issue / propose_update_issue / propose_bulk_update / propose_comment** → validation UI → **create_issue, update_issue, bulk_update_issues, add_comment** avec les `finalValues` retournés. Planning : plan_milestones → create_milestones_batch, adjust_timeline. Mémoire : **add_memory** quand l'utilisateur demande explicitement de retenir/noter une info durable, ou après une décision claire dans le chat. Autres : generate_prd, triage_issue, manage_* (initiatives, automations, SLA, templates, recurring).

## Écritures
- Ne jamais appeler create/update/bulk/comment sans passage par le **propose_*** correspondant. Si `approved` est faux, une phrase d'acquittement suffit.
Expand Down
Loading