feat: 编辑器实时探测并显示已关联虚拟环境

fix: 按系统 ANSI 代码页解码子进程输出,修复中文日志乱码
This commit is contained in:
2026-10-24 22:30:10 +08:00
parent 23565b40a4
commit dd7612bc74
20 changed files with 1508 additions and 48 deletions
+27
View File
@@ -279,6 +279,12 @@ version = "0.8.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
[[package]]
name = "core_detect"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7f8f80099a98041a3d1622845c271458a2d73e688351bf3cb999266764b81d48"
[[package]]
name = "cpufeatures"
version = "0.2.17"
@@ -485,6 +491,20 @@ dependencies = [
"serde",
]
[[package]]
name = "encoding_rs"
version = "0.8.42"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8e985e0451871ad22fb8d2b6b076e2028a502a0d3950998c2c5c0a4f9b5d9679"
dependencies = [
"cfg-if",
"core_detect",
"multiversion_no_op",
"rustversion",
"scopeguard",
"simdutf8",
]
[[package]]
name = "equivalent"
version = "1.0.3"
@@ -1208,6 +1228,12 @@ dependencies = [
"windows-sys 0.61.2",
]
[[package]]
name = "multiversion_no_op"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "743fb55ba31b18fb1ecef6bdc9aa2743314978ac084044301a7eee33fb99a20d"
[[package]]
name = "nu-ansi-term"
version = "0.50.3"
@@ -2140,6 +2166,7 @@ dependencies = [
"chrono",
"chrono-tz",
"croner",
"encoding_rs",
"futures-util",
"http-body-util",
"reqwest",
+2 -1
View File
@@ -9,6 +9,7 @@ axum = "0.8.9"
chrono = { version = "0.4.45", features = ["serde"] }
chrono-tz = "0.10.4"
croner = "4.0.1"
encoding_rs = "0.8.35"
futures-util = { version = "0.3.34", features = ["std"] }
reqwest = { version = "0.13.5", default-features = false, features = ["json", "rustls", "stream"] }
serde = { version = "1.0.229", features = ["derive"] }
@@ -25,7 +26,7 @@ url = "2.5.8"
utoipa = { version = "6.0.0", features = ["axum_extras", "chrono"] }
utoipa-axum = "0.3.0"
uuid = { version = "1.28.0", features = ["v4"] }
windows-sys = { version = "0.61.2", features = ["Win32_Foundation", "Win32_Security", "Win32_System_JobObjects"] }
windows-sys = { version = "0.61.2", features = ["Win32_Foundation", "Win32_Globalization", "Win32_Security", "Win32_System_JobObjects"] }
[dev-dependencies]
http-body-util = "0.1.5"
+32
View File
@@ -14,6 +14,9 @@ use utoipa::{OpenApi, ToSchema};
use crate::{
cron::{self, CronPreviewRequest, CronPreviewResponse},
error::{ApiError, ApiJson, ApiResult, ErrorResponse},
python_env::{
self, InterpreterKind, VenvProbeRequest, VenvProbeResponse, VenvProbeStatus, VenvSource,
},
runs::{
model::{LogChunkResponse, TaskRunListResponse, TaskRunResponse},
routes::{
@@ -107,6 +110,25 @@ pub async fn preview_cron(
Ok(Json(preview))
}
#[utoipa::path(
post,
path = "/api/v1/python/venv",
tag = "system",
request_body = VenvProbeRequest,
responses(
(status = 200, description = "Detected Python virtual environment", body = VenvProbeResponse),
(status = 400, description = "Invalid venv probe request", body = ErrorResponse)
)
)]
pub async fn probe_python_venv(
ApiJson(request): ApiJson<VenvProbeRequest>,
) -> ApiResult<Json<VenvProbeResponse>> {
tokio::task::spawn_blocking(move || python_env::probe_venv(&request))
.await
.map(Json)
.map_err(ApiError::internal)
}
#[derive(OpenApi)]
#[openapi(
info(
@@ -118,6 +140,7 @@ pub async fn preview_cron(
health,
session,
preview_cron,
probe_python_venv,
list_tasks,
create_task,
get_task,
@@ -138,6 +161,11 @@ pub async fn preview_cron(
SessionResponse,
CronPreviewRequest,
CronPreviewResponse,
VenvProbeRequest,
VenvProbeResponse,
VenvProbeStatus,
VenvSource,
InterpreterKind,
ErrorResponse,
TaskActionType,
HttpMethod,
@@ -183,6 +211,10 @@ pub fn router(state: AppState) -> Router {
.route("/api/v1/health", get(health))
.route("/api/v1/session", get(session))
.route("/api/v1/cron/preview", axum::routing::post(preview_cron))
.route(
"/api/v1/python/venv",
axum::routing::post(probe_python_venv),
)
.route("/api/v1/openapi.json", get(openapi_json))
.merge(crate::tasks::routes::router())
.merge(crate::runs::routes::router())
+374 -16
View File
@@ -6,7 +6,9 @@ use std::{
path::{Path, PathBuf},
};
use serde::{Deserialize, Serialize};
use tokio::process::Command;
use utoipa::ToSchema;
const VENV_DIRECTORY_NAMES: [&str; 3] = [".venv", "venv", "env"];
@@ -42,23 +44,8 @@ pub fn prepare_script_environment(
});
}
let working_directory = absolute_path(Path::new(working_dir))?;
let interpreter_path = resolve_interpreter(working_dir, interpreter);
let explicit_venv = configured_venv
.filter(|value| !value.trim().is_empty())
.map(|value| resolve_against(&working_directory, value))
.transpose()?;
let detected_venv = explicit_venv
.as_deref()
.map(validate_venv)
.transpose()?
.flatten()
.or_else(|| detect_venv_from_interpreter(&interpreter_path))
.or_else(|| {
is_generic_python(interpreter)
.then(|| detect_venv_from_ancestors(&working_directory))
.flatten()
});
let detected_venv = resolve_venv(working_dir, interpreter, configured_venv)?.venv;
let mut environment = BTreeMap::new();
for (key, value) in user_env {
@@ -115,6 +102,265 @@ pub fn prepare_script_environment(
})
}
/// 虚拟环境的来源,用于在界面上说明"为什么"关联到了这个环境。
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "snake_case")]
pub enum VenvSource {
/// 手动填写的 `action.venv`
Configured,
/// 由解释器路径(`...\.venv\Scripts\python.exe`)识别
Interpreter,
/// 从工作目录或上级目录探测到的 `.venv`、`venv` 或 `env`
WorkingDirectory,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ResolvedVenv {
pub venv: Option<PathBuf>,
pub source: Option<VenvSource>,
}
/// 解析任务实际会使用的虚拟环境,顺序与执行期保持一致:
/// 手动指定 → 解释器路径推断 → 工作目录向上探测。
pub fn resolve_venv(
working_dir: &str,
interpreter: &str,
configured_venv: Option<&str>,
) -> io::Result<ResolvedVenv> {
if !is_python_interpreter(interpreter) {
return Ok(ResolvedVenv {
venv: None,
source: None,
});
}
let working_directory = absolute_path(Path::new(working_dir))?;
let interpreter_path = resolve_interpreter(working_dir, interpreter);
let explicit_venv = configured_venv
.filter(|value| !value.trim().is_empty())
.map(|value| resolve_against(&working_directory, value))
.transpose()?;
if let Some(venv) = explicit_venv
.as_deref()
.map(validate_venv)
.transpose()?
.flatten()
{
return Ok(ResolvedVenv {
venv: Some(venv),
source: Some(VenvSource::Configured),
});
}
if let Some(venv) = detect_venv_from_interpreter(&interpreter_path) {
return Ok(ResolvedVenv {
venv: Some(venv),
source: Some(VenvSource::Interpreter),
});
}
if is_generic_python(interpreter)
&& let Some(venv) = detect_venv_from_ancestors(&working_directory)
{
return Ok(ResolvedVenv {
venv: Some(venv),
source: Some(VenvSource::WorkingDirectory),
});
}
Ok(ResolvedVenv {
venv: None,
source: None,
})
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "snake_case")]
pub enum InterpreterKind {
Python,
Node,
Other,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "snake_case")]
pub enum VenvProbeStatus {
/// 已关联到有效的虚拟环境
Linked,
/// 手动指定的虚拟环境无效
Invalid,
/// 没有虚拟环境,将直接使用系统解释器
Missing,
/// 解释器不是 Python,虚拟环境不适用
NotApplicable,
/// 输入不完整,暂时无法探测
Incomplete,
}
/// 编辑器在用户填写工作目录时发起的探测请求。
#[derive(Debug, Deserialize, ToSchema)]
#[serde(deny_unknown_fields)]
pub struct VenvProbeRequest {
/// 任务工作目录,对应 `action.working_dir`
pub working_dir: String,
/// 解释器或可执行程序,对应 `action.interpreter`
pub interpreter: String,
/// 手动填写的虚拟环境目录,对应 `action.venv`
#[serde(default)]
pub venv: Option<String>,
}
#[derive(Debug, Serialize, ToSchema)]
pub struct VenvProbeResponse {
pub status: VenvProbeStatus,
pub interpreter_kind: InterpreterKind,
/// 虚拟环境根目录的绝对路径
pub venv_path: Option<String>,
/// 供界面展示的解释器路径;位于工作目录内时使用相对路径
pub display_path: Option<String>,
pub source: Option<VenvSource>,
pub message: String,
}
pub fn probe_venv(request: &VenvProbeRequest) -> VenvProbeResponse {
let working_dir = request.working_dir.trim();
let interpreter = request.interpreter.trim();
let configured_venv = request
.venv
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty());
let interpreter_kind = classify_interpreter(interpreter);
if working_dir.is_empty() {
return incomplete_probe(interpreter_kind, "请先填写工作目录");
}
if interpreter.is_empty() {
return incomplete_probe(interpreter_kind, "请先填写解释器或可执行程序");
}
if !matches!(interpreter_kind, InterpreterKind::Python) {
return VenvProbeResponse {
status: VenvProbeStatus::NotApplicable,
interpreter_kind,
venv_path: None,
display_path: Some(interpreter.to_owned()),
source: None,
message: "当前解释器不是 Python,不需要虚拟环境".to_owned(),
};
}
let working_directory = match absolute_path(Path::new(working_dir)) {
Ok(path) => path,
Err(error) => {
return incomplete_probe(interpreter_kind, &format!("工作目录无效: {error}"));
}
};
if !working_directory.is_dir() {
return incomplete_probe(interpreter_kind, "工作目录不存在,暂时无法探测虚拟环境");
}
match resolve_venv(working_dir, interpreter, configured_venv) {
Ok(ResolvedVenv {
venv: Some(venv),
source,
}) => VenvProbeResponse {
status: VenvProbeStatus::Linked,
interpreter_kind,
display_path: Some(python_display_path(&venv, working_dir)),
venv_path: Some(venv.to_string_lossy().into_owned()),
source,
message: venv_source_message(source),
},
Ok(ResolvedVenv { venv: None, .. }) => VenvProbeResponse {
status: VenvProbeStatus::Missing,
interpreter_kind,
venv_path: None,
display_path: Some(interpreter.to_owned()),
source: None,
message: "未检测到虚拟环境,将直接使用系统解释器".to_owned(),
},
Err(error) => VenvProbeResponse {
status: VenvProbeStatus::Invalid,
interpreter_kind,
venv_path: None,
display_path: None,
source: None,
message: error.to_string(),
},
}
}
fn incomplete_probe(interpreter_kind: InterpreterKind, message: &str) -> VenvProbeResponse {
VenvProbeResponse {
status: VenvProbeStatus::Incomplete,
interpreter_kind,
venv_path: None,
display_path: None,
source: None,
message: message.to_owned(),
}
}
fn venv_source_message(source: Option<VenvSource>) -> String {
match source {
Some(VenvSource::Configured) => "已关联手动指定的虚拟环境".to_owned(),
Some(VenvSource::Interpreter) => "已从解释器路径识别虚拟环境".to_owned(),
Some(VenvSource::WorkingDirectory) => "已自动关联工作目录中的虚拟环境".to_owned(),
None => "已关联虚拟环境".to_owned(),
}
}
fn python_display_path(venv: &Path, working_dir: &str) -> String {
let python = venv.join("Scripts").join("python.exe");
absolute_path(Path::new(working_dir))
.ok()
.and_then(|base| relative_display(&base, &python))
.unwrap_or_else(|| python.to_string_lossy().into_owned())
}
fn relative_display(base: &Path, target: &Path) -> Option<String> {
let base_components = base.components().collect::<Vec<_>>();
let target_components = target.components().collect::<Vec<_>>();
if target_components.len() <= base_components.len() {
return None;
}
let shares_prefix = base_components
.iter()
.zip(&target_components)
.all(|(base, target)| {
base.as_os_str()
.to_string_lossy()
.eq_ignore_ascii_case(&target.as_os_str().to_string_lossy())
});
if !shares_prefix {
return None;
}
let relative = target_components[base_components.len()..]
.iter()
.map(|component| component.as_os_str())
.collect::<PathBuf>();
Some(relative.to_string_lossy().into_owned())
}
fn classify_interpreter(interpreter: &str) -> InterpreterKind {
if is_python_interpreter(interpreter) {
return InterpreterKind::Python;
}
let name = Path::new(interpreter)
.file_name()
.and_then(|name| name.to_str())
.unwrap_or(interpreter)
.to_ascii_lowercase();
let stem = name.strip_suffix(".exe").unwrap_or(&name);
if matches!(stem, "node" | "nodejs") {
InterpreterKind::Node
} else {
InterpreterKind::Other
}
}
pub fn validate_configured_venv(working_dir: &str, configured_venv: &str) -> io::Result<PathBuf> {
let working_directory = absolute_path(Path::new(working_dir))?;
let venv = resolve_against(&working_directory, configured_venv)?;
@@ -383,4 +629,116 @@ mod tests {
assert!(prepared.remove_environment.contains(&"VIRTUAL_ENV"));
assert!(prepared.remove_environment.contains(&"VIRTUAL_ENV_PROMPT"));
}
fn probe(working_dir: &str, interpreter: &str, venv: Option<&str>) -> VenvProbeResponse {
probe_venv(&VenvProbeRequest {
working_dir: working_dir.to_owned(),
interpreter: interpreter.to_owned(),
venv: venv.map(str::to_owned),
})
}
#[test]
fn probe_links_venv_detected_in_working_directory() {
let temp = TempDir::new().unwrap();
create_fake_venv(temp.path());
let response = probe(temp.path().to_str().unwrap(), "python.exe", None);
assert_eq!(response.status, VenvProbeStatus::Linked);
assert_eq!(response.source, Some(VenvSource::WorkingDirectory));
assert_eq!(response.interpreter_kind, InterpreterKind::Python);
let expected = Path::new(".venv")
.join("Scripts")
.join("python.exe")
.to_string_lossy()
.into_owned();
assert_eq!(response.display_path, Some(expected));
}
#[test]
fn probe_prefers_configured_venv_over_detection() {
let temp = TempDir::new().unwrap();
let project = temp.path().join("project");
fs::create_dir_all(&project).unwrap();
let venv = create_fake_venv(&project);
let response = probe(
project.to_str().unwrap(),
"python.exe",
Some(venv.to_str().unwrap()),
);
assert_eq!(response.status, VenvProbeStatus::Linked);
assert_eq!(response.source, Some(VenvSource::Configured));
}
#[test]
fn probe_reports_system_interpreter_when_no_venv_exists() {
let temp = TempDir::new().unwrap();
let response = probe(temp.path().to_str().unwrap(), "python.exe", None);
assert_eq!(response.status, VenvProbeStatus::Missing);
assert_eq!(response.display_path.as_deref(), Some("python.exe"));
assert!(response.venv_path.is_none());
}
#[test]
fn probe_rejects_invalid_configured_venv() {
let temp = TempDir::new().unwrap();
let missing = temp.path().join("missing-venv");
let response = probe(
temp.path().to_str().unwrap(),
"python.exe",
Some(missing.to_str().unwrap()),
);
assert_eq!(response.status, VenvProbeStatus::Invalid);
assert!(response.message.contains("pyvenv.cfg"));
}
#[test]
fn probe_marks_non_python_interpreters_as_not_applicable() {
let temp = TempDir::new().unwrap();
let working_dir = temp.path().to_str().unwrap();
let node = probe(working_dir, "node.exe", None);
assert_eq!(node.status, VenvProbeStatus::NotApplicable);
assert_eq!(node.interpreter_kind, InterpreterKind::Node);
let shell = probe(working_dir, "cmd.exe", None);
assert_eq!(shell.status, VenvProbeStatus::NotApplicable);
assert_eq!(shell.interpreter_kind, InterpreterKind::Other);
}
#[test]
fn probe_requires_existing_working_directory() {
let temp = TempDir::new().unwrap();
let empty = probe(" ", "python.exe", None);
assert_eq!(empty.status, VenvProbeStatus::Incomplete);
assert!(empty.message.contains("工作目录"));
let missing = temp.path().join("absent");
let response = probe(missing.to_str().unwrap(), "python.exe", None);
assert_eq!(response.status, VenvProbeStatus::Incomplete);
assert!(response.message.contains("工作目录"));
}
#[test]
fn probe_matches_execution_environment_for_same_input() {
let temp = TempDir::new().unwrap();
let venv = create_fake_venv(temp.path());
let working_dir = temp.path().to_str().unwrap();
let prepared =
prepare_script_environment(working_dir, "python.exe", None, &BTreeMap::new()).unwrap();
let response = probe(working_dir, "python.exe", None);
assert_eq!(prepared.virtual_env.as_deref(), Some(venv.as_path()));
assert_eq!(response.venv_path.as_deref(), Some(venv.to_str().unwrap()));
assert_eq!(response.status, VenvProbeStatus::Linked);
}
}
+135 -2
View File
@@ -10,6 +10,7 @@ use std::{
};
use chrono::Utc;
use encoding_rs::Encoding;
use futures_util::StreamExt;
use reqwest::Method;
use serde::Serialize;
@@ -39,6 +40,77 @@ const TERMINATION_WAIT_SECS: u64 = 5;
const CREATE_NO_WINDOW: u32 = 0x0800_0000;
const DEFAULT_RETENTION_DAYS: u64 = 30;
const DEFAULT_MAX_TOTAL_LOG_BYTES: u64 = 512 * 1024 * 1024;
const LOG_ENCODING_ENV: &str = "TASK_PILOT_LOG_ENCODING";
/// 子进程输出的解码方式。
///
/// `TASK_PILOT_LOG_ENCODING=auto`(默认)逐行先按 UTF-8 严格解码,失败时回退到系统
/// ANSI 代码页;填写编码标签(`gbk`、`utf-8`、`windows-1251` 等)则始终按该编码解码。
#[derive(Clone, Copy, Debug)]
pub struct OutputEncoding {
forced: Option<&'static Encoding>,
fallback: &'static Encoding,
}
impl OutputEncoding {
pub fn from_env() -> Self {
let fallback = ansi_code_page_encoding();
let forced = std::env::var(LOG_ENCODING_ENV).ok().and_then(|value| {
let label = value.trim().to_ascii_lowercase();
if label.is_empty() || label == "auto" {
None
} else {
Encoding::for_label(label.as_bytes())
}
});
Self { forced, fallback }
}
pub fn decode(&self, bytes: &[u8]) -> String {
if let Some(encoding) = self.forced {
let (text, _) = encoding.decode_without_bom_handling(bytes);
return text.into_owned();
}
match std::str::from_utf8(bytes) {
Ok(text) => text.to_owned(),
Err(_) => {
let (text, _) = self.fallback.decode_without_bom_handling(bytes);
text.into_owned()
}
}
}
/// 实际使用的编码名,写入运行日志便于排查乱码。
pub fn name(&self) -> &'static str {
self.forced.unwrap_or(self.fallback).name()
}
}
#[cfg(windows)]
fn ansi_code_page_encoding() -> &'static Encoding {
let code_page = unsafe { windows_sys::Win32::Globalization::GetACP() };
code_page_encoding(code_page)
}
#[cfg(not(windows))]
fn ansi_code_page_encoding() -> &'static Encoding {
encoding_rs::UTF_8
}
fn code_page_encoding(code_page: u32) -> &'static Encoding {
let label = match code_page {
936 => "gbk".to_owned(),
950 => "big5".to_owned(),
932 => "shift_jis".to_owned(),
949 => "euc-kr".to_owned(),
65001 => "utf-8".to_owned(),
other => format!("windows-{other}"),
};
Encoding::for_label(label.as_bytes()).unwrap_or(encoding_rs::UTF_8)
}
#[derive(Clone, Debug)]
pub struct LogRetentionConfig {
@@ -92,6 +164,7 @@ struct ExecutorInner {
http: reqwest::Client,
log_config: LogRetentionConfig,
cleanup_lock: AsyncMutex<()>,
output_encoding: OutputEncoding,
}
struct RunningHandle {
@@ -107,6 +180,7 @@ struct LogSink {
run_id: i64,
events: broadcast::Sender<ExecutionEvent>,
secrets: Arc<Vec<String>>,
encoding: OutputEncoding,
}
struct ScriptOutcome {
@@ -124,6 +198,8 @@ enum ProcessEnd {
impl TaskExecutor {
pub fn new(pool: SqlitePool, logs_dir: PathBuf, log_config: LogRetentionConfig) -> Self {
let (events, _) = broadcast::channel(512);
let output_encoding = OutputEncoding::from_env();
tracing::info!(encoding = output_encoding.name(), "task output decoding");
Self {
inner: Arc::new(ExecutorInner {
pool,
@@ -138,6 +214,7 @@ impl TaskExecutor {
.unwrap_or_default(),
log_config,
cleanup_lock: AsyncMutex::new(()),
output_encoding,
}),
}
}
@@ -353,6 +430,7 @@ impl TaskExecutor {
run_id,
self.inner.events.clone(),
secrets,
self.inner.output_encoding,
)
.await
{
@@ -517,6 +595,7 @@ impl TaskExecutor {
run_id,
self.inner.events.clone(),
sensitive_values(&task.action),
self.inner.output_encoding,
)
.await
{
@@ -638,7 +717,7 @@ impl TaskExecutor {
None => break,
}
}
let body_text = String::from_utf8_lossy(&response_body);
let body_text = self.inner.output_encoding.decode(&response_body);
if !body_text.is_empty() {
sink.http(&body_text).await;
}
@@ -697,6 +776,7 @@ impl LogSink {
run_id: i64,
events: broadcast::Sender<ExecutionEvent>,
secrets: Vec<String>,
encoding: OutputEncoding,
) -> std::io::Result<Self> {
if let Some(parent) = path.parent() {
tokio::fs::create_dir_all(parent).await?;
@@ -721,9 +801,14 @@ impl LogSink {
run_id,
events,
secrets: Arc::new(secrets),
encoding,
})
}
fn decode_output(&self, bytes: &[u8]) -> String {
self.encoding.decode(bytes)
}
async fn append(&self, text: &str) {
let redacted = redact_text(text, &self.secrets);
let bytes = redacted.as_bytes();
@@ -783,7 +868,7 @@ where
match reader.read_until(b'\n', &mut line).await {
Ok(0) => break,
Ok(_) => {
let text = String::from_utf8_lossy(&line);
let text = sink.decode_output(&line);
sink.append(&format!("[{stream}] {text}")).await;
}
Err(error) => {
@@ -957,6 +1042,54 @@ mod tests {
assert!(!is_sensitive_name("PYTHONUNBUFFERED"));
}
fn auto_encoding(fallback: &'static Encoding) -> OutputEncoding {
OutputEncoding {
forced: None,
fallback,
}
}
#[test]
fn decodes_utf8_output_without_touching_it() {
let encoding = auto_encoding(encoding_rs::GBK);
assert_eq!(encoding.decode("中文输出".as_bytes()), "中文输出");
}
#[test]
fn decodes_gbk_output_through_the_system_fallback() {
let encoding = auto_encoding(encoding_rs::GBK);
// "中文" 的 GBK 编码
let gbk = [0xD6, 0xD0, 0xCE, 0xC4];
assert_eq!(encoding.decode(&gbk), "中文");
}
#[test]
fn forced_encoding_skips_utf8_sniffing() {
let encoding = OutputEncoding {
forced: Some(encoding_rs::GBK),
fallback: encoding_rs::UTF_8,
};
let gbk = [0xD6, 0xD0, 0xCE, 0xC4];
assert_eq!(encoding.decode(&gbk), "中文");
}
#[test]
fn undecodable_bytes_still_produce_text() {
let encoding = auto_encoding(encoding_rs::UTF_8);
assert!(encoding.decode(&[0xFF, 0xFE, 0xFF]).contains('\u{fffd}'));
}
#[test]
fn maps_windows_code_pages_to_encodings() {
assert_eq!(code_page_encoding(936).name(), "GBK");
assert_eq!(code_page_encoding(65001).name(), "UTF-8");
assert_eq!(code_page_encoding(1252).name(), "windows-1252");
}
#[test]
fn cleanup_enforces_total_log_capacity() {
let temp = tempfile::TempDir::new().unwrap();
+85 -3
View File
@@ -155,13 +155,43 @@ pub async fn get_run_logs(
.await
.map_err(ApiError::internal)?;
let terminal = run.status.is_terminal();
// 读取期间任务可能仍在追加日志,用最新的文件长度判断是否真的读到了末尾。
let current_length = tokio::fs::metadata(&path)
.await
.map_err(ApiError::internal)?
.len();
let at_end_of_file = offset + bytes.len() as u64 >= current_length;
let (content, consumed) = decode_log_chunk(&bytes, at_end_of_file);
let complete = terminal && offset + consumed as u64 >= current_length;
Ok(Json(LogChunkResponse {
content: String::from_utf8_lossy(&bytes).into_owned(),
next_offset: offset + bytes.len() as u64,
complete: run.status.is_terminal(),
content,
next_offset: offset + consumed as u64,
complete,
}))
}
/// 日志文件始终以 UTF-8 写入,但按 64 KiB 切块时可能切开多字节字符。
/// 返回解码后的文本和实际消费的字节数;未读到文件末尾时保留不完整字符给下一次读取。
fn decode_log_chunk(bytes: &[u8], at_end_of_file: bool) -> (String, usize) {
match std::str::from_utf8(bytes) {
Ok(text) => (text.to_owned(), bytes.len()),
Err(error) if error.error_len().is_none() && !at_end_of_file => {
let boundary = error.valid_up_to();
if boundary > 0 {
let text = std::str::from_utf8(&bytes[..boundary])
.unwrap_or_default()
.to_owned();
return (text, boundary);
}
// 整个块都落在字符中间时不要空转,直接按损失方式消费。
(String::from_utf8_lossy(bytes).into_owned(), bytes.len())
}
Err(_) => (String::from_utf8_lossy(bytes).into_owned(), bytes.len()),
}
}
#[utoipa::path(
post,
path = "/api/v1/runs/{run_id}/cancel",
@@ -263,3 +293,55 @@ pub async fn execution_events(
Sse::new(events).keep_alive(KeepAlive::new().interval(std::time::Duration::from_secs(15)))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn keeps_complete_utf8_chunks_untouched() {
let text = "中文日志";
let (content, consumed) = decode_log_chunk(text.as_bytes(), false);
assert_eq!(content, text);
assert_eq!(consumed, text.len());
}
#[test]
fn defers_a_split_character_until_the_next_read() {
let bytes = [b'a', b'b', b'c', 0xE4];
let (content, consumed) = decode_log_chunk(&bytes, false);
assert_eq!(content, "abc");
assert_eq!(consumed, 3);
}
#[test]
fn flushes_a_split_character_at_the_end_of_the_file() {
let bytes = [b'a', b'b', b'c', 0xE4];
let (content, consumed) = decode_log_chunk(&bytes, true);
assert!(content.starts_with("abc"));
assert!(content.contains('\u{fffd}'));
assert_eq!(consumed, bytes.len());
}
#[test]
fn never_stalls_when_the_whole_chunk_is_mid_character() {
let bytes = [0xE4, 0xB8];
let (_, consumed) = decode_log_chunk(&bytes, false);
assert_eq!(consumed, bytes.len());
}
#[test]
fn replaces_truly_invalid_bytes() {
let (content, consumed) = decode_log_chunk(&[0xFF], false);
assert_eq!(consumed, 1);
assert!(content.contains('\u{fffd}'));
}
}
+181
View File
@@ -0,0 +1,181 @@
use std::{fs, path::Path};
use axum::{
Router,
body::Body,
http::{Request, StatusCode, header},
};
use http_body_util::BodyExt;
use serde_json::{Value, json};
use tempfile::TempDir;
use tower::ServiceExt;
async fn test_app() -> (Router, TempDir) {
let temp_dir = TempDir::new().expect("temporary directory should be created");
let pool = task_pilot_backend::database::connect("sqlite::memory:")
.await
.expect("test database should initialize");
let app = task_pilot_backend::app(pool, temp_dir.path().join("logs"))
.await
.expect("test app should start");
(app, temp_dir)
}
async fn fetch_token(app: &Router) -> String {
let response = app
.clone()
.oneshot(
Request::builder()
.uri("/api/v1/session")
.body(Body::empty())
.expect("request should build"),
)
.await
.expect("router should respond");
let body = response
.into_body()
.collect()
.await
.expect("session body should be readable")
.to_bytes();
let json: Value = serde_json::from_slice(&body).expect("session should be JSON");
json["token"]
.as_str()
.expect("session token should be a string")
.to_owned()
}
async fn post_probe(app: &Router, token: Option<&str>, payload: Value) -> (StatusCode, Value) {
let mut builder = Request::builder()
.method("POST")
.uri("/api/v1/python/venv")
.header(header::CONTENT_TYPE, "application/json");
if let Some(token) = token {
builder = builder.header("x-taskpilot-token", token);
}
let response = app
.clone()
.oneshot(
builder
.body(Body::from(payload.to_string()))
.expect("request should build"),
)
.await
.expect("router should respond");
let status = response.status();
let body = response
.into_body()
.collect()
.await
.expect("probe body should be readable")
.to_bytes();
(status, serde_json::from_slice(&body).unwrap_or(Value::Null))
}
fn create_fake_venv(project: &Path) {
let venv = project.join(".venv");
fs::create_dir_all(venv.join("Scripts")).expect("venv directory should be created");
fs::write(venv.join("pyvenv.cfg"), "home = C:\\Python\n")
.expect("pyvenv.cfg should be written");
fs::write(venv.join("Scripts").join("python.exe"), "")
.expect("python.exe placeholder should be written");
}
#[tokio::test]
async fn venv_probe_requires_management_token() {
let (app, _temp_dir) = test_app().await;
let (status, body) = post_probe(
&app,
None,
json!({ "working_dir": ".", "interpreter": "python.exe" }),
)
.await;
assert_eq!(status, StatusCode::FORBIDDEN);
assert_eq!(body["code"], "FORBIDDEN");
}
#[tokio::test]
async fn venv_probe_reports_linked_environment() {
let (app, temp_dir) = test_app().await;
let project = temp_dir.path().join("project");
fs::create_dir_all(&project).expect("project directory should be created");
create_fake_venv(&project);
let token = fetch_token(&app).await;
let (status, body) = post_probe(
&app,
Some(&token),
json!({
"working_dir": project.to_str().expect("project path should be UTF-8"),
"interpreter": "python.exe",
"venv": null
}),
)
.await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body["status"], "linked");
assert_eq!(body["interpreter_kind"], "python");
assert_eq!(body["source"], "working_directory");
let expected_python = Path::new(".venv")
.join("Scripts")
.join("python.exe")
.to_string_lossy()
.into_owned();
assert_eq!(body["display_path"], expected_python);
assert_eq!(
body["venv_path"],
project.join(".venv").to_string_lossy().into_owned()
);
}
#[tokio::test]
async fn venv_probe_skips_non_python_interpreters() {
let (app, temp_dir) = test_app().await;
let token = fetch_token(&app).await;
let (status, body) = post_probe(
&app,
Some(&token),
json!({
"working_dir": temp_dir.path().to_str().expect("temp path should be UTF-8"),
"interpreter": "node.exe"
}),
)
.await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body["status"], "not_applicable");
assert_eq!(body["interpreter_kind"], "node");
assert_eq!(body["display_path"], "node.exe");
}
#[tokio::test]
async fn venv_probe_reports_missing_working_directory() {
let (app, temp_dir) = test_app().await;
let token = fetch_token(&app).await;
let absent = temp_dir.path().join("absent");
let (status, body) = post_probe(
&app,
Some(&token),
json!({
"working_dir": absent.to_str().expect("absent path should be UTF-8"),
"interpreter": "python.exe"
}),
)
.await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body["status"], "incomplete");
assert!(
body["message"]
.as_str()
.expect("message should be a string")
.contains("工作目录")
);
}
+102
View File
@@ -719,3 +719,105 @@ async fn scheduler_restores_enabled_tasks_after_restart() {
second_state.shutdown();
second_pool.close().await;
}
async fn run_command_task(app: &TestApp, id: &str, args: Vec<&str>) -> i64 {
let working_dir = app._temp_dir.path().to_string_lossy().into_owned();
let task = script_task(id, &working_dir, "cmd.exe", "/C", args, "0 0 2 * * *", 10);
let create_response = send_json(&app.router, Method::POST, "/api/v1/tasks", Some(task)).await;
assert_eq!(create_response.status(), StatusCode::CREATED);
let run_response = send_json(
&app.router,
Method::POST,
&format!("/api/v1/tasks/{id}/run"),
None,
)
.await;
assert_eq!(run_response.status(), StatusCode::ACCEPTED);
response_json(run_response).await["id"]
.as_i64()
.expect("run id should be an integer")
}
async fn read_log_chunk(app: &TestApp, run_id: i64, offset: u64) -> Value {
let response = send_json(
&app.router,
Method::GET,
&format!("/api/v1/runs/{run_id}/logs?offset={offset}"),
None,
)
.await;
assert_eq!(response.status(), StatusCode::OK);
response_json(response).await
}
#[tokio::test]
async fn gbk_encoded_output_is_decoded_to_utf8() {
let app = test_app().await;
// 固定的 GBK 字节只在中日韩等使用该代码页的系统上才有意义。
let encoding = task_pilot_backend::runs::executor::OutputEncoding::from_env();
if encoding.name() != "GBK" {
return;
}
let payload = app._temp_dir.path().join("gbk.txt");
// "中文" 的 GBK 编码
std::fs::write(&payload, [0xD6, 0xD0, 0xCE, 0xC4]).expect("GBK fixture should be written");
let run_id = run_command_task(&app, "gbk_task", vec!["type", "gbk.txt"]).await;
let completed = wait_for_run(&app.router, run_id).await;
assert_eq!(completed["status"], "succeeded");
let log = read_log_chunk(&app, run_id, 0).await;
let content = log["content"].as_str().expect("log content should be text");
assert!(
content.contains("中文"),
"unexpected log content: {content}"
);
assert!(!content.contains('\u{fffd}'));
app.state.shutdown();
}
#[tokio::test]
async fn long_utf8_log_is_reassembled_across_chunk_boundaries() {
let app = test_app().await;
let payload = app._temp_dir.path().join("long.txt");
let body = format!("{}\n尾巴\n", "中".repeat(30_000));
std::fs::write(&payload, body.as_bytes()).expect("UTF-8 fixture should be written");
let run_id = run_command_task(&app, "long_log_task", vec!["type", "long.txt"]).await;
let completed = wait_for_run(&app.router, run_id).await;
assert_eq!(completed["status"], "succeeded");
let first = read_log_chunk(&app, run_id, 0).await;
assert_eq!(
first["complete"], false,
"a terminal run must not report completion before the whole log is read"
);
let mut content = first["content"].as_str().unwrap_or_default().to_owned();
let mut offset = first["next_offset"]
.as_u64()
.expect("next_offset should be an integer");
for _ in 0..20 {
let chunk = read_log_chunk(&app, run_id, offset).await;
content.push_str(chunk["content"].as_str().unwrap_or_default());
offset = chunk["next_offset"]
.as_u64()
.expect("next_offset should be an integer");
if chunk["complete"].as_bool().unwrap_or(false) {
break;
}
}
assert!(
!content.contains('\u{fffd}'),
"log lost a character boundary"
);
assert!(content.contains("尾巴"));
assert_eq!(
content.len(),
usize::try_from(offset).expect("offset fits usize")
);
app.state.shutdown();
}