Lot 43 : jobs d'arrière-plan du mode Code tenus par la façade

bash_bg et bash_tail des tours de la façade ne passent plus par /tool/exec
du Go : src/tools/jobs.rs porte code_jobs.go. Refus avant lancement
(commande vide ou dangereuse, optimisation en cours), journal
.loki/jobs/<id>.log dans le dossier de la discussion, processus qui survit
au tour, shell sans le jeton interne ; relecture bornée (40 lignes par
défaut, 200 au plus), état et durée, arrêt relu aussitôt comme le Go.
paths::go_mkdir_all reproduit os.MkdirAll, erreur comprise.

Les jobs d'une discussion supprimée s'arrêtent avec elle. L'optimiseur du
Go compte ceux de la façade (POST /internal/facade/v1/bgjobs) avant de
mesurer ; façade injoignable : les siens seulement.

Parité : journal de 24 étapes tiré du Go (parity_jobs_test.go →
tests/parity_jobs/jobs.json), rejoué par tests/jobs.rs ; un tour de la
façade qui lance un job (tests/lot6.rs) ; TestFacadeJobsArrierePlan.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016tcGfJWefW5kUGoC2wL8cv
This commit is contained in:
Claude committed 2026-10-16 05:42:32 +00:00
1 parent f66d1b7621
commit 7288db632e
16 files changed
+1132 -4

No files matched your search

+1
View File
@@ -1323,6 +1323,7 @@ impl ChatShared {
tasks: None,
memory: None,
web: Default::default(),
jobs: crate::tools::jobs::global(),
go_version: Default::default(),
}
}
+31
View File
@@ -349,6 +349,8 @@ pub struct ChatShared {
pub memory: Option<Arc<crate::memory::Memory>>,
/// Les outils web (lot 42) : cache des pages, Crawl4AI, DuckDuckGo.
pub web: Arc<crate::webtools::Web>,
/// Les jobs d'arrière-plan du mode Code (lot 43).
pub jobs: Arc<crate::tools::jobs::BgJobs>,
/// La version du Go, annoncée aux serveurs MCP à une reconnexion ; lue
/// une fois.
pub go_version: tokio::sync::OnceCell<String>,
@@ -707,6 +709,35 @@ impl TurnHost for GoTurnHost {
..Default::default()
};
}
// Jobs d'arrière-plan du mode Code (lot 43), dans le dossier de la
// discussion ; contexte indisponible : le Go s'en charge.
if name == "bash_tail" {
return ToolOutcome {
result: self.shared.jobs.bash_tail(args).await,
..Default::default()
};
}
if name == "bash_bg" {
let st = crate::tools::GoToolState {
store: self.shared.store.clone(),
caps: self.caps.clone(),
};
match crate::tools::ToolState::context(&st).await {
Ok(ctx) => {
let busy = crate::backends::tune_busy(&self.shared.store).await;
let result = self
.shared
.jobs
.bash_bg(&ctx.cwd, &self.compact.main_conv, args, busy)
.await;
return ToolOutcome {
result,
..Default::default()
};
}
Err(e) => eprintln!("[outils] contexte indisponible, bash_bg par le Go : {e}"),
}
}
// Outils task_*, quand la façade tient les tâches (lot 37).
if crate::tasks::tools::is_tool(name)
&& let Some(tasks) = &self.shared.tasks
+2
View File
@@ -65,6 +65,8 @@ pub struct GoSessionHooks(pub Store);
impl SessionHooks for GoSessionHooks {
async fn dropped(&self, id: &str) {
// Ses jobs d'arrière-plan s'arrêtent avec elle (`stopConvJobs`).
crate::tools::jobs::global().stop_conv(id);
let body = serde_json::json!({ "id": id });
if let Err(e) = self
.0
+7
View File
@@ -71,6 +71,7 @@ pub fn is_internal(path: &str) -> bool {
| "/internal/facade/v1/chat/prewarm/cancel"
| "/internal/facade/v1/perf/record"
| "/internal/facade/v1/perf/snapshot"
| "/internal/facade/v1/bgjobs"
)
}
@@ -136,6 +137,12 @@ pub async fn serve_internal(chat: &Arc<ChatService>, req: Request<Incoming>) ->
},
// L'anneau, de la plus ancienne à la plus récente (poids de
// l'optimiseur).
// Jobs d'arrière-plan encore en vie (lot 43) : l'optimiseur du Go
// refuse de mesurer à côté.
"/internal/facade/v1/bgjobs" => send_json(
StatusCode::OK,
&json!({ "running": chat.shared.jobs.running() }),
),
"/internal/facade/v1/perf/snapshot" => send_json(
StatusCode::OK,
&json!({ "recs": chat.shared.perf.snapshot() }),
+49
View File
@@ -176,6 +176,55 @@ fn mkdir_all(dir: &str) -> io::Result<()> {
b.create(dir)
}
/// `os.MkdirAll(p, 0o755)` du Go, erreur comprise : elle nomme l'élément du
/// chemin qui a échoué (un parent qui est un fichier, par exemple).
pub fn go_mkdir_all(p: &Path) -> Result<(), String> {
let err = |e: &io::Error| os_err("mkdir", &p.to_string_lossy(), e);
match std::fs::metadata(p) {
Ok(m) if m.is_dir() => return Ok(()),
Ok(_) => return Err(err(&io::Error::from_raw_os_error(libc_enotdir()))),
Err(_) => {}
}
let s = p.to_string_lossy();
let b = s.as_bytes();
let mut i = b.len();
while i > 0 && b[i - 1] == b'/' {
i -= 1;
}
let mut j = i;
while j > 0 && b[j - 1] != b'/' {
j -= 1;
}
if j > 1 {
go_mkdir_all(Path::new(&s[..j - 1]))?;
}
let mut d = std::fs::DirBuilder::new();
#[cfg(unix)]
{
use std::os::unix::fs::DirBuilderExt;
d.mode(0o755);
}
if let Err(e) = d.create(p) {
if std::fs::symlink_metadata(p).is_ok_and(|m| m.is_dir()) {
return Ok(());
}
return Err(err(&e));
}
Ok(())
}
/// ENOTDIR de la plateforme.
fn libc_enotdir() -> i32 {
#[cfg(unix)]
{
libc::ENOTDIR
}
#[cfg(not(unix))]
{
267 // ERROR_DIRECTORY
}
}
/// Équivalent de `os.CreateTemp(dir, ".probe-*")` suivi de sa suppression.
fn probe_writable(dir: &str) -> bool {
for _ in 0..16 {
+245
View File
@@ -0,0 +1,245 @@
//! Commandes shell d'arrière-plan du mode Code (lot 43) — port de
//! `code_jobs.go` : `bash_bg` lance un processus détaché (serveur de dev,
//! watcher, build long) et rend la main ; `bash_tail` relit sa sortie et son
//! état, et l'arrête sur demande.
//!
//! La sortie va dans `.loki/jobs/<id>.log` du dossier de la discussion : elle
//! survit au tour, et le panneau Fichiers la montre. Les jobs d'une
//! discussion supprimée sont arrêtés ; l'optimiseur refuse de mesurer tant
//! qu'il en tourne (le Go demande leur nombre à la façade).
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant};
use serde_json::{Map, Value};
use super::policy;
use super::shell::{TOOL_MAX_OUTPUT, go_duration, shell_cmd};
use crate::system::go_lossy;
/// Un job lancé par `bash_bg`.
struct Job {
id: String,
command: String,
log_path: PathBuf,
/// Discussion du tour qui l'a lancé.
conv: String,
started: Instant,
/// `None` tant qu'il tourne, puis son code de sortie.
exit: Mutex<Option<i32>>,
kill: tokio::sync::Notify,
}
impl Job {
fn running(&self) -> bool {
self.exit
.lock()
.unwrap_or_else(|e| e.into_inner())
.is_none()
}
}
/// Les jobs d'arrière-plan de la façade.
#[derive(Default)]
pub struct BgJobs {
state: Mutex<(u64, HashMap<String, Arc<Job>>)>,
}
/// Les jobs de ce processus (les tours et la suppression des discussions).
pub fn global() -> Arc<BgJobs> {
static G: OnceLock<Arc<BgJobs>> = OnceLock::new();
G.get_or_init(Default::default).clone()
}
/// `args[k].(string)`.
fn arg_str<'a>(args: &'a Map<String, Value>, k: &str) -> &'a str {
args.get(k).and_then(Value::as_str).unwrap_or("")
}
impl BgJobs {
/// `toolBashBg` : `tune_busy`, le refus de l'optimiseur en cours s'il y
/// en a un.
pub async fn bash_bg(
&self,
cwd: &Path,
conv: &str,
args: &Map<String, Value>,
tune_busy: Option<String>,
) -> String {
let command = arg_str(args, "command");
if command.trim().is_empty() {
return "[erreur] commande vide".into();
}
if let Some(reason) = policy::dangerous_command(command) {
return policy::refused_command_result(reason);
}
// Un serveur ou un build lancé pendant une optimisation fausserait
// ses mesures.
if let Some(e) = tune_busy {
return format!("[erreur] {e}");
}
let log_dir = cwd.join(".loki").join("jobs");
if let Err(e) = crate::paths::go_mkdir_all(&log_dir) {
return format!("[erreur] {e}");
}
let id = {
let mut st = self.state.lock().unwrap_or_else(|e| e.into_inner());
st.0 += 1;
format!("job{}", st.0)
};
let log_path = log_dir.join(format!("{id}.log"));
let log = match std::fs::File::create(&log_path) {
Ok(f) => f,
Err(e) => {
return format!(
"[erreur] {}",
crate::paths::os_err("open", &log_path.to_string_lossy(), &e)
);
}
};
let err_log = match log.try_clone() {
Ok(f) => f,
Err(e) => return format!("[erreur] {e}"),
};
// Pas le contexte du tour : le job lui survit, c'est sa raison d'être.
let mut cmd = shell_cmd(command);
cmd.current_dir(cwd)
.stdin(std::process::Stdio::null())
.stdout(log)
.stderr(err_log)
.kill_on_drop(false);
let mut child = match cmd.spawn() {
Ok(c) => c,
Err(e) => {
return format!(
"[erreur] {}",
crate::paths::os_err("fork/exec", "/bin/bash", &e)
);
}
};
let pid = child.id().unwrap_or(0);
let job = Arc::new(Job {
id: id.clone(),
command: command.to_string(),
log_path,
conv: conv.to_string(),
started: Instant::now(),
exit: Mutex::new(None),
kill: tokio::sync::Notify::new(),
});
self.state
.lock()
.unwrap_or_else(|e| e.into_inner())
.1
.insert(id.clone(), job.clone());
tokio::spawn(async move {
let status = tokio::select! {
s = child.wait() => s,
_ = job.kill.notified() => {
let _ = child.start_kill();
child.wait().await
}
};
// `ExitCode()` du Go : -1 pour un processus tué par un signal ;
// 0 si l'attente elle-même a échoué.
let code = match status {
Ok(s) => s.code().unwrap_or(-1),
Err(_) => 0,
};
*job.exit.lock().unwrap_or_else(|e| e.into_inner()) = Some(code);
});
format!(
"[ok] job {id} lancé (pid {pid}) — sortie dans .loki/jobs/{id}.log, suis-la avec bash_tail"
)
}
/// `toolBashTail` : la fin du journal d'un job et son état ; `kill`
/// l'arrête d'abord.
pub async fn bash_tail(&self, args: &Map<String, Value>) -> String {
let id = arg_str(args, "id");
let job = self
.state
.lock()
.unwrap_or_else(|e| e.into_inner())
.1
.get(id)
.cloned();
let Some(job) = job else {
return format!("[erreur] job inconnu : {id}");
};
// Tué puis relu aussitôt, comme le Go : l'état peut encore dire « en
// cours » le temps que l'attente constate la fin.
if args.get("kill").and_then(Value::as_bool) == Some(true) && job.running() {
job.kill.notify_one();
}
let mut lines = 40usize;
if let Some(v) = args.get("lines").and_then(Value::as_f64) {
let n = crate::webtools::go_int(v);
if n > 0 {
lines = n.min(200) as usize;
}
}
let b = match std::fs::read(&job.log_path) {
Ok(b) => b,
Err(e) => {
return format!(
"[erreur] {}",
crate::paths::os_err("open", &job.log_path.to_string_lossy(), &e)
);
}
};
let text = go_lossy(&b);
let all: Vec<&str> = text.trim_end_matches('\n').split('\n').collect();
let all = &all[all.len().saturating_sub(lines)..];
let status = match *job.exit.lock().unwrap_or_else(|e| e.into_inner()) {
None => "en cours".to_string(),
Some(c) => format!("terminé (exit {c})"),
};
let since = Duration::from_secs((job.started.elapsed().as_millis() as u64 + 500) / 1000);
let out = format!(
"job {} — {status} — depuis {}\ncommande : {}\n",
job.id,
go_duration(since),
job.command
);
let mut tail = all.join("\n");
if tail.trim().is_empty() {
tail = "(aucune sortie pour l'instant)".into();
}
format!(
"{out}\n{}",
crate::webtools::tail_runes(&tail, TOOL_MAX_OUTPUT)
)
}
/// `bgJobsRunning` : les jobs encore en vie.
pub fn running(&self) -> usize {
self.state
.lock()
.unwrap_or_else(|e| e.into_inner())
.1
.values()
.filter(|j| j.running())
.count()
}
/// `stopConvJobs` : les jobs d'une discussion supprimée, arrêtés et
/// oubliés.
pub fn stop_conv(&self, conv: &str) {
let mut st = self.state.lock().unwrap_or_else(|e| e.into_inner());
st.1.retain(|_, j| {
if j.conv != conv {
return true;
}
if j.running() {
j.kill.notify_one();
}
false
});
}
}
#[cfg(test)]
mod tests;
+76
View File
@@ -0,0 +1,76 @@
use super::*;
fn args(kv: &[(&str, Value)]) -> Map<String, Value> {
kv.iter().map(|(k, v)| (k.to_string(), v.clone())).collect()
}
/// Optimisation en cours : refus du Go recopié ; discussion supprimée : ses
/// jobs arrêtés et oubliés, ceux des autres gardés.
#[cfg(unix)]
#[tokio::test]
async fn refus_et_discussion_supprimee() {
let tmp = tempfile::tempdir().unwrap();
let jobs = BgJobs::default();
let a = args(&[("command", Value::from("sleep 30"))]);
assert_eq!(
jobs.bash_bg(tmp.path(), "c", &a, Some("optimisation en cours".into()))
.await,
"[erreur] optimisation en cours"
);
assert!(!tmp.path().join(".loki").exists());
assert!(
jobs.bash_bg(tmp.path(), "c1", &a, None)
.await
.starts_with("[ok] job job1 lancé")
);
assert!(
jobs.bash_bg(tmp.path(), "c2", &a, None)
.await
.starts_with("[ok] job job2 lancé")
);
assert_eq!(jobs.running(), 2);
jobs.stop_conv("c1");
for _ in 0..500 {
if jobs.running() == 1 {
break;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
assert_eq!(jobs.running(), 1);
assert_eq!(
jobs.bash_tail(&args(&[("id", Value::from("job1"))])).await,
"[erreur] job inconnu : job1"
);
jobs.stop_conv("c2");
for _ in 0..500 {
if jobs.running() == 0 {
break;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
assert_eq!(jobs.running(), 0);
}
/// Dossier des journaux impossible à créer : l'erreur du Go.
#[cfg(unix)]
#[tokio::test]
async fn dossier_impossible() {
let tmp = tempfile::tempdir().unwrap();
std::fs::write(tmp.path().join(".loki"), "un fichier").unwrap();
let jobs = BgJobs::default();
let got = jobs
.bash_bg(
tmp.path(),
"c",
&args(&[("command", Value::from("true"))]),
None,
)
.await;
assert_eq!(
got,
format!(
"[erreur] mkdir {}: not a directory",
tmp.path().join(".loki").display()
)
);
}
+1
View File
@@ -16,6 +16,7 @@
pub mod diff;
pub mod files;
pub mod git;
pub mod jobs;
pub mod policy;
pub mod shell;
+1 -1
View File
@@ -140,7 +140,7 @@ fn drain<R: tokio::io::AsyncRead + Unpin + Send + 'static>(
/// La commande du shell de la plateforme. Le jeton de l'API interne n'a rien
/// à faire dans le shell de l'agent.
fn shell_cmd(command: &str) -> tokio::process::Command {
pub(crate) fn shell_cmd(command: &str) -> tokio::process::Command {
#[cfg(windows)]
let mut c = {
let mut c = tokio::process::Command::new("cmd");