diff --git a/bin/pudox-run/src/main.rs b/bin/pudox-run/src/main.rs index 98083e5..7c6e9b8 100644 --- a/bin/pudox-run/src/main.rs +++ b/bin/pudox-run/src/main.rs @@ -16,6 +16,7 @@ macro_rules! pudoxd_url { const PUDOXD_OPENAI_URL: &str = pudoxd_url!("/v1"); const PUDOXD_TASK_URL: &str = pudoxd_url!("/task"); +const PUDOXD_HISTORY_URL: &str = pudoxd_url!("/task/history"); const PUDOXD_RESULT_URL: &str = pudoxd_url!("/task/result"); const PUDOXD_STREAM_URL: &str = pudoxd_url!("/task/stream"); const PUDOXD_INFERENCE_URL: &str = pudoxd_url!("/v1/chat/completions"); @@ -89,6 +90,8 @@ async fn connect_sse(session_id: String, tx: tokio::sync::mpsc::Sender) struct Task { title: String, description: String, + history: Vec, + pending_prompts: Vec, } async fn load_task(session_id: &str) -> Result> { @@ -96,6 +99,10 @@ async fn load_task(session_id: &str) -> Result> struct TaskPayload { title: String, description: String, + #[serde(default)] + history: Vec, + #[serde(default)] + pending_prompts: Vec, } let resp = reqwest::Client::new() @@ -106,7 +113,6 @@ async fn load_task(session_id: &str) -> Result> let status = resp.status(); let body = resp.text().await?; - // tracing::info!("GET /task status={} body={}", status, body); let payload: TaskPayload = serde_json::from_str(&body) .map_err(|e| format!("GET /task parse error (status={}): {}", status, e))?; @@ -114,6 +120,8 @@ async fn load_task(session_id: &str) -> Result> Ok(Task { title: payload.title, description: payload.description, + history: payload.history, + pending_prompts: payload.pending_prompts, }) } @@ -137,17 +145,36 @@ async fn process_task( let mut messages: Vec = Vec::new(); - messages.push(ChatCompletionRequestMessage::User( - ChatCompletionRequestUserMessageArgs::default() - .content("You are a coding agent.") - .build()?, - )); - - messages.push(ChatCompletionRequestMessage::User( - ChatCompletionRequestUserMessageArgs::default() - .content(task.description.clone()) - .build()?, - )); + if task.history.is_empty() { + messages.push(ChatCompletionRequestMessage::User( + ChatCompletionRequestUserMessageArgs::default() + .content("You are a coding agent.") + .build()?, + )); + messages.push(ChatCompletionRequestMessage::User( + ChatCompletionRequestUserMessageArgs::default() + .content(task.description.clone()) + .build()?, + )); + } else { + for raw in &task.history { + let msg: ChatCompletionRequestMessage = serde_json::from_value(raw.clone()) + .map_err(|e| format!("Failed to deserialize history message: {}", e))?; + messages.push(msg); + } + for prompt in &task.pending_prompts { + messages.push(ChatCompletionRequestMessage::User( + ChatCompletionRequestUserMessageArgs::default() + .content(prompt.clone()) + .build()?, + )); + } + tracing::info!( + "Resuming session with {} history messages and {} pending prompts", + task.history.len(), + task.pending_prompts.len() + ); + } let mut tools: Vec = Vec::new(); @@ -265,6 +292,11 @@ async fn process_task( } } + tracing::info!("Saving message history ({} messages)...", messages.len()); + if let Err(e) = save_messages(session_id, &messages).await { + tracing::warn!("Failed to save message history: {}", e); + } + tracing::info!("Summarizing session ({} messages)...", messages.len()); let summary = summarize_session(&messages).await?; tracing::info!("Session summarized ({} chars)", summary.len()); @@ -672,6 +704,28 @@ async fn tool_ripgrep(arguments: &str) -> String { } } +async fn save_messages( + session_id: &str, + messages: &[ChatCompletionRequestMessage], +) -> Result<(), Box> { + let serialized: Vec = messages + .iter() + .map(|m| serde_json::to_value(m)) + .collect::>()?; + + let client = reqwest::Client::new(); + let response = client + .post(PUDOXD_HISTORY_URL) + .json(&serde_json::json!({ + "session_id": session_id, + "messages": serialized, + })) + .send() + .await?; + tracing::info!("History saved: {}", response.status()); + Ok(()) +} + async fn report_task_result(result: &TaskResult) -> Result<(), Box> { let client = reqwest::Client::new(); let response = client.post(PUDOXD_RESULT_URL).json(result).send().await?; diff --git a/bin/pudoxd/src/main.rs b/bin/pudoxd/src/main.rs index 769d70e..965b081 100644 --- a/bin/pudoxd/src/main.rs +++ b/bin/pudoxd/src/main.rs @@ -44,6 +44,7 @@ async fn main() -> Result<(), Box> { let router = Router::new() .route("/webhook", post(webhook_handler)) .route("/task", get(task_handler)) + .route("/task/history", post(task_history_handler)) .route("/task/result", post(task_result_handler)) .route("/task/stream", get(stream_handler)) .route("/v1/chat/completions", post(chat_completions_proxy)) @@ -61,6 +62,8 @@ async fn main() -> Result<(), Box> { struct TaskPayload { title: String, description: String, + history: Vec, + pending_prompts: Vec, } async fn task_handler(State(app): State, headers: axum::http::HeaderMap) -> impl IntoResponse { @@ -69,7 +72,7 @@ async fn task_handler(State(app): State, headers: axum::http::HeaderMap) -> None => return (StatusCode::BAD_REQUEST, "Missing X-Session-Id header").into_response(), }; - let prompt_context = { + let (prompt_context, messages_arc, pending_arc) = { let sessions = app.sessions.read().await; let s = sessions.get(&session_id); tracing::info!( @@ -78,7 +81,14 @@ async fn task_handler(State(app): State, headers: axum::http::HeaderMap) -> s.is_some(), s.and_then(|s| s.prompt_context.as_ref()).is_some() ); - s.and_then(|s| s.prompt_context.clone()) + match s { + Some(s) => ( + s.prompt_context.clone(), + Some(s.messages.clone()), + Some(s.pending_prompts.clone()), + ), + None => (None, None, None), + } }; let description = match prompt_context { @@ -88,9 +98,25 @@ async fn task_handler(State(app): State, headers: axum::http::HeaderMap) -> } }; + let history = match messages_arc { + Some(arc) => arc.read().await.clone(), + None => vec![], + }; + + let pending_prompts = match pending_arc { + Some(arc) => { + let mut lock = arc.lock().await; + let drained: Vec = lock.drain(..).collect(); + drained + } + None => vec![], + }; + Json(TaskPayload { title: session_id, description, + history, + pending_prompts, }) .into_response() } @@ -148,6 +174,165 @@ async fn task_result_handler( (axum::http::StatusCode::OK, "Result received").into_response() } +#[derive(serde::Deserialize)] +struct TaskHistoryPayload { + session_id: String, + messages: Vec, +} + +async fn task_history_handler( + State(app): State, + axum::Json(payload): axum::Json, +) -> impl IntoResponse { + let sessions = app.sessions.read().await; + match sessions.get(&payload.session_id) { + Some(session) => { + *session.messages.write().await = payload.messages; + tracing::info!("Saved message history for session {}", payload.session_id); + (StatusCode::OK, "History saved").into_response() + } + None => { + tracing::warn!("task/history: unknown session {}", payload.session_id); + (StatusCode::NOT_FOUND, "Session not found").into_response() + } + } +} + +async fn upsert_pull_request( + client: &reqwest::Client, + token: &str, + repo_slug: &str, + branch: &str, + title: &str, + body: Option<&str>, +) -> Option { + let auth = format!("Bearer {}", token); + let base_url = format!("https://api.github.com/repos/{}/pulls", repo_slug); + + // Try to create the PR first. + let mut create_body = serde_json::json!({ + "title": title, + "head": branch, + "base": "main", + }); + if let Some(b) = body { + create_body["body"] = serde_json::Value::String(b.to_string()); + } + + let create_resp = match client + .post(&base_url) + .header("Authorization", &auth) + .header("User-Agent", "pudox/0.1") + .header("Accept", "application/vnd.github+json") + .json(&create_body) + .send() + .await + { + Ok(r) => r, + Err(e) => { + tracing::error!("GitHub API request failed: {}", e); + return None; + } + }; + + if create_resp.status().is_success() { + return match create_resp.json::().await { + Ok(json) => { + let url = json["html_url"].as_str().unwrap_or("").to_string(); + tracing::info!("Pull request opened: {}", url); + Some(url) + } + Err(e) => { + tracing::error!("Failed to parse PR create response: {}", e); + None + } + }; + } + + // 422 means the PR already exists — fetch it and update instead. + if create_resp.status().as_u16() != 422 { + let status = create_resp.status(); + let text = create_resp.text().await.unwrap_or_default(); + tracing::error!("GitHub PR creation failed {}: {}", status, text); + return None; + } + + tracing::info!("PR already exists for branch {}, updating instead", branch); + + // Find the existing PR by head branch. + let list_url = format!("{}?head={}&state=open", base_url, branch); + let list_resp = match client + .get(&list_url) + .header("Authorization", &auth) + .header("User-Agent", "pudox/0.1") + .header("Accept", "application/vnd.github+json") + .send() + .await + { + Ok(r) => r, + Err(e) => { + tracing::error!("GitHub API list PRs failed: {}", e); + return None; + } + }; + + let prs: serde_json::Value = match list_resp.json().await { + Ok(v) => v, + Err(e) => { + tracing::error!("Failed to parse PR list response: {}", e); + return None; + } + }; + + let pr_number = match prs[0]["number"].as_u64() { + Some(n) => n, + None => { + tracing::error!("No open PR found for branch {}", branch); + return None; + } + }; + + let patch_url = format!("https://api.github.com/repos/{}/pulls/{}", repo_slug, pr_number); + let mut patch_body = serde_json::json!({ "title": title }); + if let Some(b) = body { + patch_body["body"] = serde_json::Value::String(b.to_string()); + } + + match client + .patch(&patch_url) + .header("Authorization", &auth) + .header("User-Agent", "pudox/0.1") + .header("Accept", "application/vnd.github+json") + .json(&patch_body) + .send() + .await + { + Ok(resp) if resp.status().is_success() => { + match resp.json::().await { + Ok(json) => { + let url = json["html_url"].as_str().unwrap_or("").to_string(); + tracing::info!("Pull request updated: {}", url); + Some(url) + } + Err(e) => { + tracing::error!("Failed to parse PR update response: {}", e); + None + } + } + } + Ok(resp) => { + let status = resp.status(); + let text = resp.text().await.unwrap_or_default(); + tracing::error!("GitHub PR update failed {}: {}", status, text); + None + } + Err(e) => { + tracing::error!("GitHub API PATCH failed: {}", e); + None + } + } +} + async fn handle_task_result( linear: crate::linear::Linear, session_id: String, @@ -240,56 +425,22 @@ async fn handle_task_result( let github_token = std::env::var("GITHUB_TOKEN").unwrap_or_default(); let pr_url = if repo_slug.is_empty() || github_token.is_empty() { tracing::warn!( - "Skipping PR creation: repo_slug={:?}, token_present={}", + "Skipping PR upsert: repo_slug={:?}, token_present={}", repo_slug, !github_token.is_empty() ); None } else { - let api_url = format!("https://api.github.com/repos/{}/pulls", repo_slug); let client = reqwest::Client::new(); - let mut body = serde_json::json!({ - "title": payload.commit_title, - "head": branch_name, - "base": "main", - }); - if let Some(b) = &payload.pull_request_description { - body["body"] = serde_json::Value::String(b.clone()); - } - - match client - .post(&api_url) - .header("Authorization", format!("Bearer {}", github_token)) - .header("User-Agent", "pudox/0.1") - .header("Accept", "application/vnd.github+json") - .json(&body) - .send() - .await - { - Ok(resp) if resp.status().is_success() => { - match resp.json::().await { - Ok(json) => { - let url = json["html_url"].as_str().unwrap_or("").to_string(); - tracing::info!("Pull request opened: {}", url); - Some(url) - } - Err(e) => { - tracing::error!("Failed to parse PR response: {}", e); - None - } - } - } - Ok(resp) => { - let status = resp.status(); - let text = resp.text().await.unwrap_or_default(); - tracing::error!("GitHub PR creation failed {}: {}", status, text); - None - } - Err(e) => { - tracing::error!("GitHub API request failed: {}", e); - None - } - } + upsert_pull_request( + &client, + &github_token, + &repo_slug, + &branch_name, + &payload.commit_title, + payload.pull_request_description.as_deref(), + ) + .await }; // ── 3. Link PR to Linear issue ───────────────────────────────────────────── @@ -485,6 +636,8 @@ async fn webhook_handler(State(app): State, body: String) -> &'static str { .map(|c| c.email.clone()); let (tx, rx) = mpsc::channel::(32); let (sse_tx, _) = tokio::sync::broadcast::channel::(64); + let messages = Arc::new(tokio::sync::RwLock::new(Vec::::new())); + let pending_prompts = Arc::new(tokio::sync::Mutex::new(Vec::::new())); let handle = tokio::spawn(run_session( app.sessions.clone(), @@ -493,6 +646,7 @@ async fn webhook_handler(State(app): State, body: String) -> &'static str { session_id.clone(), app.linear.clone(), sse_tx.clone(), + pending_prompts.clone(), )); app.sessions.write().await.insert( @@ -505,6 +659,8 @@ async fn webhook_handler(State(app): State, body: String) -> &'static str { creator_name, creator_email, sse_tx, + messages, + pending_prompts, }, ); diff --git a/bin/pudoxd/src/session.rs b/bin/pudoxd/src/session.rs index 0ca9669..a70f9dd 100644 --- a/bin/pudoxd/src/session.rs +++ b/bin/pudoxd/src/session.rs @@ -20,11 +20,14 @@ pub struct Session { pub creator_name: Option, pub creator_email: Option, pub sse_tx: tokio::sync::broadcast::Sender, + pub messages: Arc>>, + pub pending_prompts: Arc>>, } pub(crate) struct SandboxContext { pub(crate) dir: std::path::PathBuf, pub(crate) process: tokio::process::Child, + pub(crate) dev_env: HashMap, } async fn resolve_git_url( @@ -142,36 +145,43 @@ async fn init( return; } - let dir_path = dir.clone(); - let url = repo_url.clone(); - let token = std::env::var("GITHUB_TOKEN").unwrap_or_default(); - let clone_result = tokio::task::spawn_blocking(move || { - let mut callbacks = git2::RemoteCallbacks::new(); - callbacks.credentials(move |_url, _username, _allowed| { - git2::Cred::userpass_plaintext("x-access-token", &token) - }); - let mut fetch_opts = git2::FetchOptions::new(); - fetch_opts.remote_callbacks(callbacks); - let mut builder = git2::build::RepoBuilder::new(); - builder.fetch_options(fetch_opts); - builder.clone(&url, &dir_path) - }) - .await; - - match clone_result { - Ok(Ok(_)) => tracing::info!("Cloned {} for session {}", repo_url, session_id), - Ok(Err(e)) => { - tracing::error!( - "Failed to clone {} for session {}: {}", - repo_url, - session_id, - e - ); - return; - } - Err(e) => { - tracing::error!("Clone task panicked for session {}: {}", session_id, e); - return; + if dir.join(".git").exists() { + tracing::info!( + "Workspace already exists for session {}, skipping clone", + session_id + ); + } else { + let dir_path = dir.clone(); + let url = repo_url.clone(); + let token = std::env::var("GITHUB_TOKEN").unwrap_or_default(); + let clone_result = tokio::task::spawn_blocking(move || { + let mut callbacks = git2::RemoteCallbacks::new(); + callbacks.credentials(move |_url, _username, _allowed| { + git2::Cred::userpass_plaintext("x-access-token", &token) + }); + let mut fetch_opts = git2::FetchOptions::new(); + fetch_opts.remote_callbacks(callbacks); + let mut builder = git2::build::RepoBuilder::new(); + builder.fetch_options(fetch_opts); + builder.clone(&url, &dir_path) + }) + .await; + + match clone_result { + Ok(Ok(_)) => tracing::info!("Cloned {} for session {}", repo_url, session_id), + Ok(Err(e)) => { + tracing::error!( + "Failed to clone {} for session {}: {}", + repo_url, + session_id, + e + ); + return; + } + Err(e) => { + tracing::error!("Clone task panicked for session {}: {}", session_id, e); + return; + } } } @@ -252,21 +262,23 @@ async fn init( .send(SessionMessage::SandboxReady(SandboxContext { dir, process, + dev_env, })) .await; } pub async fn run_session( - sessions: Arc>>, + _sessions: Arc>>, tx: mpsc::Sender, mut rx: mpsc::Receiver, session_id: String, linear: Linear, sse_tx: tokio::sync::broadcast::Sender, + pending_prompts: Arc>>, ) { - // After SandboxReady, we only need the workspace dir. The child process is - // handed off to a watcher task that will send ProcessExited. let mut workspace_dir: Option = None; + let mut dev_env: HashMap = HashMap::new(); + let mut process_running = false; while let Some(msg) = rx.recv().await { match msg { @@ -278,10 +290,14 @@ pub async fn run_session( SessionMessage::SandboxReady(ctx) => { tracing::info!("Sandbox ready for session {}", session_id); - // Destructure so each part can be moved independently. - let SandboxContext { dir, mut process } = ctx; + let SandboxContext { + dir, + mut process, + dev_env: sandbox_dev_env, + } = ctx; + + dev_env = sandbox_dev_env; - // Spawn a watcher: when the child exits, notify the session loop. let tx2 = tx.clone(); tokio::spawn(async move { let _ = process.wait().await; @@ -289,50 +305,64 @@ pub async fn run_session( }); workspace_dir = Some(dir); + process_running = true; } SessionMessage::Prompted(body) => { tracing::info!("Session {} received prompt: {}", session_id, body); - let data = serde_json::json!({ "type": "prompt", "body": body }).to_string(); - let _ = sse_tx.send(data); // ignored if no subscribers connected yet - } - SessionMessage::ProcessExited => { - tracing::info!("pudox-run exited for session {}", session_id); - - let res = linear - .send(serde_json::json!({ - "query": r#" - mutation AgentActivityCreate($input: AgentActivityCreateInput!) { - agentActivityCreate(input: $input) { - success - } - } - "#, - "variables": { - "input": { - "agentSessionId": session_id, - "content": { - "type": "response", - "body": "Agent process exited." - } - } + if process_running { + let data = serde_json::json!({ "type": "prompt", "body": body }).to_string(); + let _ = sse_tx.send(data); + } else { + // Buffer the prompt and re-spawn pudox-run. + pending_prompts.lock().await.push(body); + + let Some(ref dir) = workspace_dir else { + tracing::warn!( + "Session {} prompted but workspace not ready yet", + session_id + ); + continue; + }; + + let pudox_run = std::env::current_exe() + .ok() + .and_then(|p| p.parent().map(|d| d.join("pudox-run"))) + .unwrap_or_else(|| "pudox-run".into()); + + match tokio::process::Command::new(&pudox_run) + .arg(&session_id) + .current_dir(dir) + .envs(&dev_env) + .spawn() + { + Ok(process) => { + tracing::info!("Re-spawned pudox-run for session {}", session_id); + let _ = tx + .send(SessionMessage::SandboxReady(SandboxContext { + dir: dir.clone(), + process, + dev_env: dev_env.clone(), + })) + .await; } - })) - .await; - - match res { - Ok(r) => tracing::info!("Agent session completed: {}", r), - Err(e) => tracing::error!("Failed to complete agent session: {}", e), + Err(e) => { + tracing::error!( + "Failed to re-spawn pudox-run for session {}: {}", + session_id, + e + ); + } + } } + } - break; + SessionMessage::ProcessExited => { + tracing::info!("pudox-run exited for session {}", session_id); + process_running = false; + // Linear was already notified by handle_task_result; no action needed here. } } } - - drop(workspace_dir); - - sessions.write().await.remove(&session_id); - tracing::info!("Session {} removed from state", session_id); }