Compare commits

..
11 Commits
Author SHA1 Message Date
semantic-release-bot 6680795ce7 chore(release): 1.15.2 [skip ci]
## [1.15.2](https://github.com/asepharyana/zesdex/compare/v1.15.1...v1.15.2) (2026-07-17)

### Bug Fixes

* correct jitter range to ±25% and fix abort.rs doc — review findings ([9b6e51d](https://github.com/asepharyana/zesdex/commit/9b6e51dc677cc4b798532f1e63335785f2f7bbbf))
2026-07-17 20:21:33 +00:00
asepharyana 9b6e51dc67 fix: correct jitter range to ±25% and fix abort.rs doc — review findings 2026-07-18 03:18:27 +07:00
asepharyana 40524f930f refactor: use mark_dirty() helper di input.rs, DRY 22 lokasi 2026-07-18 03:18:27 +07:00
asepharyana 086cb86f0d refactor: extract is_aborted helpers, DRY 16 call sites 2026-07-18 03:18:27 +07:00
asepharyana ff47bffe0c refactor: reuse is_auth_error from provider.rs, DRY backend retry logic 2026-07-18 03:18:27 +07:00
asepharyana 431c8d3b89 refactor: unify 3 backoff implementations into shared helper 2026-07-18 03:18:27 +07:00
asepharyanaandClaude Opus 4.8 e32501ee59 refactor: extract session_id extraction helper, DRY auth.rs
Extract duplicated session ID extraction + validation logic from
SessionAuthMiddleware::call() and require_session() into two shared
helper functions: extract_session_id and validate_and_build_identity.

Removes ~60 lines of duplicated code while preserving behavior:
- Both call sites now rely on the same extraction/validation path
- User-Agent default remains empty string (existing behavior unchanged)
- Error response format (401 with header/validation messages) unchanged

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-18 03:18:27 +07:00
asepharyanaandClaude Opus 4.8 7d8487cefb refactor: consolidate #![allow(clippy::cast_*)] to crate roots
Add the cast-allow block to zesdex-entities/src/lib.rs and
zesdex-utils/src/lib.rs (which lacked it), then remove from
65 sub-files across all 8 crates. Build and all 223 tests
continue to pass.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-18 03:18:27 +07:00
asepharyanaandClaude Opus 4.8 4cd38c9291 refactor: extract write_json_atomic helper, DRY 8 call sites
Move the crash-safe write-then-rename pattern into
zesdex-utils::write_json_atomic and apply it across:

- zesdex-cms: app_config_repo, conversation_repo, settings_repo
- zesdex-iam: oauth_repo, session_repo
- zesdex-entities: Conversation::save_conversation, Session::save

Excluded (non-JSON format):
- rewind_blob_repo (binary blob)
- memory_repo (markdown + frontmatter, not JSON)
- session_lock (PID string, not JSON)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-18 03:18:27 +07:00
asepharyanaandClaude Opus 4.8 9a6ab62562 refactor: DRY cleanup — extract shared helpers, remove duplication across tools, LSP, overlays, and runtime
Eliminate ~500 lines of duplicate code across 31 files by extracting
shared functions, helpers, and consolidating repeated patterns.

Highlights:
- Toast helpers (toast_info/success/warning/error) on AppStateRest
- push_event() helper for turn-event queue (19 callers consolidated)
- log_write_edit_tool() shared fn (turn.rs + engine.rs ~50 lines saved)
- resolve_api_key() shared fn (spawn.rs + provider.rs)
- LSP call_positional() helper on LspClient
- lsp_cursor_params() shared schema for 4 tool files
-overlay_block() helper for consistent overlay title/border styling
- cycle_selected_index(), path_not_found/a_directory() helpers
- mark_dirty(), save_settings() on AppStateRest
- Remove redundant Err(e) => Err(e) arms in LSP tools
- Consolidate generate_workspace_tree (turn.rs → workspace.rs)
- Simplify background-review wrapper args in auto/mod.rs

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-18 03:18:27 +07:00
asepharyana b02754acd2 refactor: implement retry logic with exponential backoff and jitter for subagent and provider calls 2026-07-18 03:18:27 +07:00
106 changed files with 2161 additions and 1519 deletions
+7
View File
@@ -1,3 +1,10 @@
## [1.15.2](https://github.com/asepharyana/zesdex/compare/v1.15.1...v1.15.2) (2026-07-17)
### Bug Fixes
* correct jitter range to ±25% and fix abort.rs doc — review findings ([9b6e51d](https://github.com/asepharyana/zesdex/commit/9b6e51dc677cc4b798532f1e63335785f2f7bbbf))
## [1.15.1](https://github.com/asepharyana/zesdex/compare/v1.15.0...v1.15.1) (2026-07-17)
# [1.15.0](https://github.com/asepharyana/zesdex/compare/v1.14.0...v1.15.0) (2026-07-17)
Generated
+9 -8
View File
@@ -4671,7 +4671,7 @@ dependencies = [
[[package]]
name = "zesdex-backend"
version = "1.15.1"
version = "1.15.2"
dependencies = [
"anyhow",
"base64",
@@ -4721,7 +4721,7 @@ dependencies = [
[[package]]
name = "zesdex-cms"
version = "1.15.1"
version = "1.15.2"
dependencies = [
"anyhow",
"chrono",
@@ -4737,7 +4737,7 @@ dependencies = [
[[package]]
name = "zesdex-entities"
version = "1.15.1"
version = "1.15.2"
dependencies = [
"anyhow",
"base64",
@@ -4752,11 +4752,12 @@ dependencies = [
"tracing",
"url",
"uuid",
"zesdex-utils",
]
[[package]]
name = "zesdex-iam"
version = "1.15.1"
version = "1.15.2"
dependencies = [
"anyhow",
"base64",
@@ -4777,7 +4778,7 @@ dependencies = [
[[package]]
name = "zesdex-infra"
version = "1.15.1"
version = "1.15.2"
dependencies = [
"anyhow",
"argon2",
@@ -4800,7 +4801,7 @@ dependencies = [
[[package]]
name = "zesdex-ipc"
version = "1.15.1"
version = "1.15.2"
dependencies = [
"anyhow",
"serde",
@@ -4811,7 +4812,7 @@ dependencies = [
[[package]]
name = "zesdex-middleware"
version = "1.15.1"
version = "1.15.2"
dependencies = [
"anyhow",
"axum",
@@ -4826,7 +4827,7 @@ dependencies = [
[[package]]
name = "zesdex-utils"
version = "1.15.1"
version = "1.15.2"
dependencies = [
"anyhow",
"base64",
+1 -1
View File
@@ -12,7 +12,7 @@ members = [
]
[workspace.package]
version = "1.15.1"
version = "1.15.2"
edition = "2021"
authors = ["asepharyana <superaseph@gmail.com>"]
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Global registry of running background bash jobs, and control operations
//! (output polling, kill) exposed to the rest of the app.
//!
+42 -30
View File
@@ -303,24 +303,35 @@ impl LspClient {
)
}
/// Call a textDocument/positional method (hover, completion, definition, references).
///
/// Builds the standard `{ textDocument: { uri }, position: { line, character } }` body
/// and delegates to `self.call`. `extra` is merged into the body when present (used by
/// `references` to include the `context` block).
fn call_positional(
&mut self,
method: &str,
uri: &str,
line: u32,
character: u32,
extra: Option<serde_json::Value>,
) -> anyhow::Result<Value> {
let mut body = json!({
"textDocument": { "uri": uri },
"position": { "line": line, "character": character },
});
if let Some(ref extra) = extra {
merge_json(&mut body, extra);
}
self.call(method, &body)
}
pub fn hover(&mut self, uri: &str, line: u32, character: u32) -> anyhow::Result<Value> {
self.call(
"textDocument/hover",
&json!({
"textDocument": { "uri": uri },
"position": { "line": line, "character": character }
}),
)
self.call_positional("textDocument/hover", uri, line, character, None)
}
pub fn completion(&mut self, uri: &str, line: u32, character: u32) -> anyhow::Result<Value> {
self.call(
"textDocument/completion",
&json!({
"textDocument": { "uri": uri },
"position": { "line": line, "character": character }
}),
)
self.call_positional("textDocument/completion", uri, line, character, None)
}
pub fn goto_definition(
@@ -329,25 +340,13 @@ impl LspClient {
line: u32,
character: u32,
) -> anyhow::Result<Value> {
self.call(
"textDocument/definition",
&json!({
"textDocument": { "uri": uri },
"position": { "line": line, "character": character }
}),
)
self.call_positional("textDocument/definition", uri, line, character, None)
}
pub fn references(&mut self, uri: &str, line: u32, character: u32) -> anyhow::Result<Value> {
self.call(
"textDocument/references",
&json!({
"textDocument": { "uri": uri },
"position": { "line": line, "character": character },
"context": {
"includeDeclaration": true
}
}),
self.call_positional(
"textDocument/references", uri, line, character,
Some(json!({"context": { "includeDeclaration": true }})),
)
}
@@ -384,6 +383,19 @@ impl Drop for LspClient {
}
}
/// Merge the fields of `b` into the object `a` (mutating `a` in place).
///
/// Used by `LspClient::call_positional` to layer extra fields (e.g. `context`)
/// onto the standard positional-query body. When `a` is not an object or `b`
/// is not an object this is a no-op.
fn merge_json(a: &mut serde_json::Value, b: &serde_json::Value) {
if let (Some(map), Some(extra)) = (a.as_object_mut(), b.as_object()) {
for (k, v) in extra {
map.insert(k.clone(), v.clone());
}
}
}
pub fn path_to_lsp_uri(path: &str) -> String {
file_path_to_uri(path)
}
+1
View File
@@ -10,4 +10,5 @@ pub mod review;
pub mod runtime;
pub mod state;
pub mod subagent;
pub mod util;
pub mod workflow;
+1 -10
View File
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Effort mode: cycles the agent's reasoning effort level, which scales the
//! LLM's temperature and `max_tokens` for subsequent turns.
use crate::app::state::rest::AppStateRest;
@@ -47,9 +41,6 @@ pub fn cycle_effort(state: &mut AppStateRest) {
let current = current_effort(state);
state.misc.effort_level = (current + 1) % EFFORT_LEVELS.len();
let label = current_effort_str(state);
state.push_toast(crate::app::state::types::Toast::new(
crate::app::state::types::ToastKind::Info,
format!("Effort: {label}"),
));
state.toast_info(format!("Effort: {label}"));
state.dirty = true;
}
+19
View File
@@ -11,3 +11,22 @@ pub mod quit_confirm;
pub mod rewind;
pub mod settings;
pub mod todo;
/// Cycle `current` in the range `[0, len)`.
///
/// * `forward = true` — increment (wrap at len)
/// * `forward = false` — decrement (wrap at 0), saturating at 0 when len is 0
///
/// Return: `0` when `len == 0`, otherwise the wrapped index.
pub fn cycle_selected_index(current: usize, len: usize, forward: bool) -> usize {
if len == 0 {
return 0;
}
if forward {
(current + 1) % len
} else if current == 0 {
len.saturating_sub(1)
} else {
current - 1
}
}
+7 -34
View File
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Rewind mode: restores a file to a pre-edit snapshot stored in the
//! session's `SQLite` blob store.
use crate::app::state::rest::AppStateRest;
@@ -27,10 +21,7 @@ pub fn rewind_to(state: &mut AppStateRest, index: usize) {
let conn = match open_session_db(&state.session_dir) {
Ok(c) => c,
Err(e) => {
state.push_toast(crate::app::state::types::Toast::new(
crate::app::state::types::ToastKind::Error,
format!("Failed to open session DB: {e}"),
));
state.toast_error(format!("Failed to open session DB: {e}"));
state.dirty = true;
return;
}
@@ -39,20 +30,14 @@ pub fn rewind_to(state: &mut AppStateRest, index: usize) {
let keys = match crate::model::msglog::blobs::list_blob_keys(&conn, &state.session_id) {
Ok(k) => k,
Err(e) => {
state.push_toast(crate::app::state::types::Toast::new(
crate::app::state::types::ToastKind::Error,
format!("Failed to list snapshots: {e}"),
));
state.toast_error(format!("Failed to list snapshots: {e}"));
state.dirty = true;
return;
}
};
if keys.is_empty() || index >= keys.len() {
state.push_toast(crate::app::state::types::Toast::new(
crate::app::state::types::ToastKind::Warning,
"No snapshot available at that index".to_string(),
));
state.toast_warning("No snapshot available at that index".to_string());
state.dirty = true;
return;
}
@@ -62,18 +47,12 @@ pub fn rewind_to(state: &mut AppStateRest, index: usize) {
{
Ok(Some(b)) => b,
Ok(None) => {
state.push_toast(crate::app::state::types::Toast::new(
crate::app::state::types::ToastKind::Error,
"Snapshot data not found".to_string(),
));
state.toast_error("Snapshot data not found".to_string());
state.dirty = true;
return;
}
Err(e) => {
state.push_toast(crate::app::state::types::Toast::new(
crate::app::state::types::ToastKind::Error,
format!("Failed to retrieve snapshot: {e}"),
));
state.toast_error(format!("Failed to retrieve snapshot: {e}"));
state.dirty = true;
return;
}
@@ -87,16 +66,10 @@ pub fn rewind_to(state: &mut AppStateRest, index: usize) {
match std::fs::write(&restore_path, &bytes) {
Ok(()) => {
state.push_toast(crate::app::state::types::Toast::new(
crate::app::state::types::ToastKind::Success,
format!("Restored {} from snapshot", restore_path.display()),
));
state.toast_success(format!("Restored {} from snapshot", restore_path.display()));
}
Err(e) => {
state.push_toast(crate::app::state::types::Toast::new(
crate::app::state::types::ToastKind::Error,
format!("Failed to write restored file: {e}"),
));
state.toast_error(format!("Failed to write restored file: {e}"));
}
}
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Build/test probing: running a verification command and capturing its
//! pass/fail/timeout outcome for the review subagent.
@@ -16,8 +16,6 @@
//! running turns on plain OS threads (rather than blocking the main loop)
//! keeps the TUI responsive while the LLM streams.
#![allow(clippy::cast_possible_truncation, clippy::cast_sign_loss, clippy::cast_precision_loss, clippy::cast_possible_wrap)]
mod handlers;
mod io;
mod memory;
@@ -57,12 +57,9 @@ pub(super) fn spawn_turn(state: &AppStateRest) {
if messages.is_empty() {
return;
}
let mut api_key = state
.settings
.api_keys
.get(&state.settings.provider)
.cloned()
.unwrap_or_default();
let mut api_key = crate::service::provider::resolve_api_key(
&state.settings, &state.app_config,
);
let model = state.settings.model.clone();
let base_url = state
.app_config
@@ -95,16 +92,6 @@ pub(super) fn spawn_turn(state: &AppStateRest) {
}
return;
}
if api_key.is_empty() {
if let Some(provider_cfg) = state.app_config.providers.get(&state.settings.provider) {
api_key = provider_cfg
.api_key_env
.as_ref()
.and_then(|env| std::env::var(env).ok())
.or_else(|| provider_cfg.default_api_key.clone())
.unwrap_or_default();
}
}
if api_key.is_empty() {
api_key = crate::service::provider::DEFAULT_API_KEY.to_string();
}
@@ -271,6 +271,19 @@ pub(super) fn handle_tick(state: &mut AppStateRest) {
}
TurnEvent::StreamDone(msg) => {
state.misc.thinking = false;
// Replace the partial streaming transcript with the complete
// message content. In the normal streaming path this is a
// no-op (the accumulated tokens already match), but when the
// non-streaming fallback fires the response is a completely
// new generation — the partial SSE text must be overwritten.
if let Some(content) = &msg.content {
if let Some(last) = state.transcript_cache.messages.last_mut() {
if last.role == Role::Assistant {
last.content.clone_from(content);
state.transcript_cache.dirty = true;
}
}
}
if let Some(ref mut rt) = state.session_runtime {
rt.push_message(msg);
}
@@ -3,18 +3,17 @@
//! messages, and manages auto-retry for unfinished tasks.
//!
//! Also contains the smaller helpers that the loop depends on:
//! `execute_one_tool`, `generate_workspace_tree`, `build_memory_section`,
//! and `archive_message`.
//! `execute_one_tool`, `build_memory_section`, and `archive_message`.
use std::collections::VecDeque;
use std::fmt::Write;
use sha2::Digest;
use zesdex_cms::domain::repository::EditLogRepository;
use crate::app::guard::Verdict;
use crate::app::runtime::context::tokens::count_tokens;
use crate::app::runtime::push_event;
use crate::app::state::runtime::TurnEvent;
use zesdex_cms::domain::repository::EditLogRepository;
use zesdex_cms::domain::repository::MemoryRepository;
use crate::dto::chat::message::ChatMessage;
@@ -66,17 +65,15 @@ pub(super) fn run_agent_turn(
const MAX_TODO_RETRIES: usize = 5;
let mut msgs = messages.to_vec();
let mut edited_paths: Vec<String> = Vec::new();
let initial_edits = zesdex_cms::infrastructure::persistence::edit_log_repo::JsonlEditLogRepository::new()
.open(&tc.edit_log_session_dir)
.map(|el| el.len())
.unwrap_or(0);
let initial_edit_log = zesdex_cms::infrastructure::persistence::edit_log_repo::JsonlEditLogRepository::new()
.open(&tc.edit_log_session_dir).ok();
let mut inline_reviews_count: usize = 0;
let mut prev_shaped = false;
// Build system prompt components once and cache them for the entire turn
// instead of regenerating on every loop iteration (which walks the full
// workspace tree and reads all memory files each time).
let tree_info = generate_workspace_tree(&tc.workspace_roots);
let tree_info = crate::app::subagent::workspace::generate_workspace_tree(&tc.workspace_roots);
let memory_section = build_memory_section(&tc.ctx.memory_dir);
let system_text = format!(
"{}\n\n{}\n\n{}{}",
@@ -142,12 +139,10 @@ pub(super) fn run_agent_turn(
"[hive-mind] the Hive stirs — Core Intelligence compiling a cognitive cycle plan"
);
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::SystemNote {
kind: "pipeline".to_string(),
message: HIVE_MIND_KICKOFF_NOTE.to_string(),
});
}
push_event(&events_q, TurnEvent::SystemNote {
kind: "pipeline".to_string(),
message: HIVE_MIND_KICKOFF_NOTE.to_string(),
});
let pipeline_abort = Some(tc.abort_flag.clone());
@@ -194,8 +189,13 @@ pub(super) fn run_agent_turn(
let planner_prompt_chars = system_msg.content.as_deref().map_or(0, str::len)
+ user_msg.content.as_deref().map_or(0, str::len);
let planner_result =
tc.client.chat_with_tools_non_streaming(&[system_msg, user_msg], None);
let planner_result = tc.client.chat_with_tools_non_streaming(
&[system_msg, user_msg],
None,
None,
None,
Some(&tc.abort_flag),
);
let pipeline_result = match planner_result {
Ok((reply, usage_opt)) => {
let (mut tok_in, mut tok_out) = usage_opt.unwrap_or((0, 0));
@@ -206,12 +206,10 @@ pub(super) fn run_agent_turn(
let response_chars = reply.content.as_deref().map_or(0, str::len);
tok_out = (response_chars / 4).max(1) as u64;
}
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::Usage {
tokens_in: tok_in,
tokens_out: tok_out,
});
}
push_event(&events_q, TurnEvent::Usage {
tokens_in: tok_in,
tokens_out: tok_out,
});
let reply_text = reply.content.as_deref().unwrap_or("").trim();
let clean_json = if reply_text.starts_with("```") {
let mut lines = reply_text.lines();
@@ -238,15 +236,13 @@ pub(super) fn run_agent_turn(
.collect::<Vec<String>>()
.join(", ");
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::SystemNote {
kind: "pipeline".to_string(),
message: format!(
"The Hive compiled {} cycle(s) — {cycle_desc}. Deploying nodes...",
plan.cycles.len()
),
});
}
push_event(&events_q, TurnEvent::SystemNote {
kind: "pipeline".to_string(),
message: format!(
"The Hive compiled {} cycle(s) — {cycle_desc}. Deploying nodes...",
plan.cycles.len()
),
});
crate::app::workflow::hive_mind::run_hive_mind(
user_request,
@@ -283,20 +279,16 @@ pub(super) fn run_agent_turn(
archive_message(tc.db.as_ref(), &tc.session_id, &pipeline_msg);
msgs.push(pipeline_msg);
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::SystemNote {
kind: "pipeline".to_string(),
message:
"The Hive's convergence is complete. Core Intelligence reviewing consensus for LO..."
.to_string(),
});
}
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::SystemNote {
kind: "hive_mind_converged".to_string(),
message: String::new(),
});
}
push_event(&events_q, TurnEvent::SystemNote {
kind: "pipeline".to_string(),
message:
"The Hive's convergence is complete. Core Intelligence reviewing consensus for LO..."
.to_string(),
});
push_event(&events_q, TurnEvent::SystemNote {
kind: "hive_mind_converged".to_string(),
message: String::new(),
});
}
Err(e) => {
tracing::warn!("[hive-mind] convergence fractured: {}", e);
@@ -314,13 +306,9 @@ pub(super) fn run_agent_turn(
// Check abort after pipeline completes, before entering main loop.
// This catches the case where the user pressed Esc during the pipeline
// phase, which previously ran unchecked for minutes at a time.
if tc
.abort_flag
.load(std::sync::atomic::Ordering::SeqCst)
if crate::app::util::abort::is_aborted_direct(&tc.abort_flag)
{
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::Error("Generation aborted by user".to_string()));
}
push_event(&events_q, TurnEvent::Error("Generation aborted by user".to_string()));
return Ok(());
}
@@ -336,9 +324,7 @@ pub(super) fn run_agent_turn(
// Skip message compaction if abort was requested — the non-streaming
// LLM call for summarization would block without checking abort_flag.
let wire_msgs = if !tc
.abort_flag
.load(std::sync::atomic::Ordering::SeqCst)
let wire_msgs = if !crate::app::util::abort::is_aborted_direct(&tc.abort_flag)
&& crate::app::runtime::context::shaping::should_shape(
token_estimate,
max_wire_tokens,
@@ -357,9 +343,7 @@ pub(super) fn run_agent_turn(
// Dispatch the compacted messages to the main thread so the local session history
// is permanently compacted and doesn't trigger shaping again immediately on next turn.
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::Compacted(compacted.clone()));
}
push_event(&events_q, TurnEvent::Compacted(compacted.clone()));
// Also update our local `msgs` variable so the rest of the loop operates on the compacted version
msgs.clone_from(&compacted);
@@ -383,7 +367,7 @@ pub(super) fn run_agent_turn(
Some(tc.temperature),
tc.max_tokens,
|event| -> bool {
if tc.abort_flag.load(std::sync::atomic::Ordering::SeqCst) {
if crate::app::util::abort::is_aborted_direct(&tc.abort_flag) {
return false;
}
if let Ok(mut q) = events_q.lock() {
@@ -424,28 +408,25 @@ pub(super) fn run_agent_turn(
}
true
},
Some(&tc.abort_flag),
);
if reasoning_started && !reasoning_ended {
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::StreamToken(
"\n</think>\n\n".to_string(),
));
}
push_event(&events_q, TurnEvent::StreamToken(
"\n</think>\n\n".to_string(),
));
}
let (response, final_usage) = match result {
Ok((msg, u)) => (msg, u.or(usage)),
Err(e) => {
// If abort was requested, return immediately.
if tc.abort_flag.load(std::sync::atomic::Ordering::SeqCst)
if crate::app::util::abort::is_aborted_direct(&tc.abort_flag)
|| e.to_string().contains("aborted")
{
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::Error(
"Generation aborted by user".to_string(),
));
}
push_event(&events_q, TurnEvent::Error(
"Generation aborted by user".to_string(),
));
return Ok(());
}
// Streaming-only: no non-streaming fallback.
@@ -472,14 +453,12 @@ pub(super) fn run_agent_turn(
Edit todo.md manually or ask me to focus on specific items.",
);
}
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::SystemNote {
kind: "task_retry".to_string(),
message: format!(
"Network/API error: {api_err}. Auto-retrying to finish tasks... (retry {todo_retry_count}/{MAX_TODO_RETRIES})"
),
});
}
push_event(&events_q, TurnEvent::SystemNote {
kind: "task_retry".to_string(),
message: format!(
"Network/API error: {api_err}. Auto-retrying to finish tasks... (retry {todo_retry_count}/{MAX_TODO_RETRIES})"
),
});
std::thread::sleep(std::time::Duration::from_secs(5));
continue;
}
@@ -500,12 +479,10 @@ pub(super) fn run_agent_turn(
let response_chars = response.content.as_deref().map_or(0, str::len);
tok_out = (response_chars / 4).max(1) as u64;
}
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::Usage {
tokens_in: tok_in,
tokens_out: tok_out,
});
}
push_event(&events_q, TurnEvent::Usage {
tokens_in: tok_in,
tokens_out: tok_out,
});
let has_tool_calls = response.tool_calls.is_some()
&& response.tool_calls.as_ref().is_some_and(|tc| !tc.is_empty());
@@ -571,15 +548,11 @@ pub(super) fn run_agent_turn(
});
for (tool_call, tool_name, args, output, is_error, is_edit) in results_vec {
if tc
.abort_flag
.load(std::sync::atomic::Ordering::SeqCst)
if crate::app::util::abort::is_aborted_direct(&tc.abort_flag)
{
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::Error(
"Turn aborted by user".to_string(),
));
}
push_event(&events_q, TurnEvent::Error(
"Turn aborted by user".to_string(),
));
return Ok(());
}
@@ -648,17 +621,13 @@ pub(super) fn run_agent_turn(
.and_then(|v| v.as_str())
.map(std::string::ToString::to_string);
{
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::ToolResult {
tool_call_id: tool_call.id.clone(),
tool_name: tool_name.clone(),
output: output.clone(),
is_error,
path: tool_path,
});
}
}
push_event(&events_q, TurnEvent::ToolResult {
tool_call_id: tool_call.id.clone(),
tool_name: tool_name.clone(),
output: output.clone(),
is_error,
path: tool_path,
});
let tool_msg =
ChatMessage::tool_result(tool_call.id.clone(), output);
@@ -668,12 +637,10 @@ pub(super) fn run_agent_turn(
} else {
if !content.is_empty() {
archive_message(tc.db.as_ref(), &tc.session_id, &response);
if let Ok(mut q) = events_q.lock() {
if stream_started {
q.push_back(TurnEvent::StreamDone(response.clone()));
} else {
q.push_back(TurnEvent::AssistantMessage(response.clone()));
}
if stream_started {
push_event(&events_q, TurnEvent::StreamDone(response.clone()));
} else {
push_event(&events_q, TurnEvent::AssistantMessage(response.clone()));
}
}
@@ -691,12 +658,10 @@ pub(super) fn run_agent_turn(
if has_unfinished {
todo_retry_count += 1;
if todo_retry_count > MAX_TODO_RETRIES {
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::SystemNote {
kind: "task_retry".to_string(),
message: format!("Giving up after {MAX_TODO_RETRIES} retries — some todo items remain unfinished. Edit todo.md manually or ask again."),
});
}
push_event(&events_q, TurnEvent::SystemNote {
kind: "task_retry".to_string(),
message: format!("Giving up after {MAX_TODO_RETRIES} retries — some todo items remain unfinished. Edit todo.md manually or ask again."),
});
break;
}
let sys_text = format!("You stopped, but you still have unfinished tasks in todo.md (marked with '- [ ]'). You MUST continue working and use tools to finish them, or edit todo.md to mark them as done if they are finished. (Retry {todo_retry_count}/{MAX_TODO_RETRIES})");
@@ -704,12 +669,10 @@ pub(super) fn run_agent_turn(
let msg = ChatMessage::system(sys_text);
archive_message(tc.db.as_ref(), &tc.session_id, &msg);
msgs.push(msg);
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::SystemNote {
kind: "task_retry".to_string(),
message: sys_text_clone,
});
}
push_event(&events_q, TurnEvent::SystemNote {
kind: "task_retry".to_string(),
message: sys_text_clone,
});
continue;
}
@@ -717,49 +680,52 @@ pub(super) fn run_agent_turn(
}
}
let el = zesdex_cms::infrastructure::persistence::edit_log_repo::JsonlEditLogRepository::new()
.open(&tc.edit_log_session_dir)
.unwrap_or_else(|_| zesdex_cms::domain::edit_log::EditLog::new());
let final_edits = el.len();
let total_edits_this_turn = final_edits.saturating_sub(initial_edits);
let total_edits_this_turn = initial_edit_log.as_ref().and_then(|initial_el| {
let initial_count = initial_el.len();
zesdex_cms::infrastructure::persistence::edit_log_repo::JsonlEditLogRepository::new()
.open(&tc.edit_log_session_dir)
.ok()
.map(|final_el| {
let count = final_el.len().saturating_sub(initial_count);
(count, initial_count, final_el)
})
});
if total_edits_this_turn > 0 {
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::SystemNote {
if let Some((total_edits_this_turn, prev_edits, el)) = &total_edits_this_turn {
if *total_edits_this_turn > 0 {
push_event(&events_q, TurnEvent::SystemNote {
kind: "edits".to_string(),
message: total_edits_this_turn.to_string(),
});
}
// Collect edited paths from the new edit log entries
let mut bg_paths = Vec::new();
for entry in el.entries.iter().skip(initial_edits) {
bg_paths.push(entry.path.clone());
}
bg_paths.sort();
bg_paths.dedup();
// Collect edited paths from the new edit log entries
let mut bg_paths = Vec::new();
for entry in el.entries.iter().skip(*prev_edits) {
bg_paths.push(entry.path.clone());
}
bg_paths.sort();
bg_paths.dedup();
// ── Background auto-subagents ──
if !bg_paths.is_empty() {
let bg_session_dir = tc.edit_log_session_dir.clone();
let bg_workspaces = tc.workspace_roots.clone();
let bg_events = events_q.clone();
let bg_abort = tc.abort_flag.clone();
std::thread::spawn(move || {
crate::app::subagent::auto::spawn_all_background(
&bg_paths,
&bg_session_dir,
&bg_workspaces,
&bg_events,
bg_abort,
);
});
// ── Background auto-subagents ──
if !bg_paths.is_empty() {
let bg_session_dir = tc.edit_log_session_dir.clone();
let bg_workspaces = tc.workspace_roots.clone();
let bg_events = events_q.clone();
let bg_abort = tc.abort_flag.clone();
std::thread::spawn(move || {
crate::app::subagent::auto::spawn_all_background(
&bg_paths,
&bg_session_dir,
&bg_workspaces,
&bg_events,
bg_abort,
);
});
}
}
}
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::Done);
}
push_event(&events_q, TurnEvent::Done);
Ok(())
}
@@ -820,55 +786,9 @@ fn execute_one_tool(
}
let result = tool.run(ctx, args)?;
if name == "write" || name == "edit" {
let reason = args
.get("reason")
.and_then(|v| v.as_str())
.unwrap_or("unnamed");
let path = args
.get("path")
.and_then(|v| v.as_str())
.unwrap_or("unknown");
let content_sha256 = {
let content =
args.get("content").or_else(|| args.get("new"));
let hash = sha2::Sha256::digest(
content
.and_then(|v| v.as_str())
.unwrap_or("")
.as_bytes(),
);
hex::encode(hash)
};
let bytes_delta = if name == "write" {
args.get("content")
.and_then(|v| v.as_str())
.map_or(0, |s| s.len() as i64)
} else {
let old = args
.get("old")
.and_then(|v| v.as_str())
.unwrap_or("");
let new = args
.get("new")
.and_then(|v| v.as_str())
.unwrap_or("");
(new.len() as i64 - old.len() as i64).abs()
};
let entry = zesdex_cms::domain::edit_log::EditLogEntry {
ts: chrono::Utc::now().timestamp_millis(),
tool: name.to_string(),
path: path.to_string(),
reason: reason.to_string(),
content_sha256,
bytes_delta,
origin: ctx.origin.tag(),
session_id: sess.id.to_string(),
};
let repo =
zesdex_cms::infrastructure::persistence::edit_log_repo::JsonlEditLogRepository::new();
if let Ok(mut el) = repo.open(sess.dir) {
let _ = repo.append(sess.dir, &mut el, entry);
}
crate::tool::log_write_edit_tool(
args, name, &ctx.origin.tag(), sess.dir, sess.id,
);
}
return Ok(result);
}
@@ -876,44 +796,6 @@ fn execute_one_tool(
anyhow::bail!("tool not found: {name}")
}
/// Build an ASCII tree of the workspace directory structure for the
/// system prompt, so the LLM can see the file layout.
///
/// Flow: for each root, walk using `ignore::WalkBuilder` (respecting
/// `.gitignore` and hidden files) → prefix `[DIR]` for directories →
/// truncate after 1000 entries.
///
/// Return: a formatted string with one entry per line.
fn generate_workspace_tree(roots: &[std::path::PathBuf]) -> String {
let mut out = String::new();
out.push_str("Current Workspace Directory Structure:\n");
for root in roots {
writeln!(out, "Root: {}", root.display()).unwrap();
let walker = ignore::WalkBuilder::new(root)
.hidden(true)
.git_ignore(true)
.build();
let mut count = 0;
for entry in walker.flatten() {
let path = entry.path();
if let Ok(rel) = path.strip_prefix(root) {
if rel.as_os_str().is_empty() {
continue;
}
let is_dir = entry.file_type().is_some_and(|ft| ft.is_dir());
let prefix = if is_dir { "[DIR] " } else { " " };
writeln!(out, " {}{}", prefix, rel.display()).unwrap();
count += 1;
if count > 1000 {
out.push_str(" ... (truncated)\n");
break;
}
}
}
}
out
}
/// Load all memory entries from `memory_dir` and format them as a compact
/// section appended to the system prompt, so the AI is always aware of
/// stored lessons and project knowledge.
@@ -315,7 +315,7 @@ pub fn shape_messages(
let mut result: Option<String> = None;
let mut last_err: Option<anyhow::Error> = None;
for attempt in 0..2 {
match llm.chat_with_tools_non_streaming(&req_msgs, None) {
match llm.chat_with_tools_non_streaming(&req_msgs, None, None, None, abort_flag) {
Ok(resp) => {
if let Some(content) = resp.0.content {
result = Some(format!(
@@ -1,6 +1,26 @@
//! Runtime layer: action dispatch, slash commands, short-send handling,
//! and the LLM streaming pipeline.
use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
use super::state::runtime::TurnEvent;
pub mod actions;
pub mod action_dispatch;
pub mod context;
pub mod stream;
/// Acquire the mutex on a turn-events queue and push one event onto it.
///
/// Silently ignores a poisoned mutex so callers never have to handle lock
/// errors inline. Used by the 20+ locations in `actions/turn.rs` that push
/// events and want to skip the boilerplate.
pub fn push_event(
q: &Arc<Mutex<VecDeque<TurnEvent>>>,
event: TurnEvent,
) {
if let Ok(mut guard) = q.lock() {
guard.push_back(event);
}
}
+37 -1
View File
@@ -347,10 +347,35 @@ impl AppStateRest {
self.dirty = true;
}
/// Mark the app state as dirty, triggering a TUI re-render on the next frame.
pub fn mark_dirty(&mut self) {
self.dirty = true;
}
/// Queue a toast notification for display and mark the app dirty.
pub fn push_toast(&mut self, toast: Toast) {
self.misc.push_toast(toast);
self.dirty = true;
self.mark_dirty();
}
/// Push an info toast with the given message.
pub fn toast_info(&mut self, msg: impl Into<String>) {
self.push_toast(Toast::new(super::types::ToastKind::Info, msg.into()));
}
/// Push a success toast with the given message.
pub fn toast_success(&mut self, msg: impl Into<String>) {
self.push_toast(Toast::new(super::types::ToastKind::Success, msg.into()));
}
/// Push a warning toast with the given message.
pub fn toast_warning(&mut self, msg: impl Into<String>) {
self.push_toast(Toast::new(super::types::ToastKind::Warning, msg.into()));
}
/// Push an error toast with the given message.
pub fn toast_error(&mut self, msg: impl Into<String>) {
self.push_toast(Toast::new(super::types::ToastKind::Error, msg.into()));
}
/// Resolve the base directory that stores this session (grandparent of
@@ -385,6 +410,17 @@ impl AppStateRest {
)
}
/// Persist the current settings to the store and swallow any error.
///
/// Inline usage of `JsonSettingsRepository::new().save(...)` was
/// duplicated twice in `controller/input.rs` — this helper centralises
/// the call site.
pub fn save_settings(&self) {
use zesdex_cms::domain::repository::SettingsRepository;
let _ = zesdex_cms::infrastructure::persistence::settings_repo::JsonSettingsRepository::new()
.save(&self.store_base_dir(), &self.settings);
}
/// Build a `ToolCtx` for tool calls originating from the main agent.
pub fn tool_ctx(&self) -> crate::tool::ToolCtx {
self.tool_ctx_for(Origin::Main)
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Shared small state types: toasts, overlays, the transcript cache,
//! tool execution model, and call origin tags.
use serde::{Deserialize, Serialize};
@@ -248,16 +248,24 @@ fn spawn_background_review(
});
}
/// Collect the trailing arguments shared by all background-review spawners.
fn review_args<'a>(
file_paths: &'a [String],
session_dir: &'a Path,
workspaces: &'a [std::path::PathBuf],
turn_events: &'a Arc<Mutex<VecDeque<TurnEvent>>>,
abort_flag: Arc<AtomicBool>,
) -> (Vec<String>, std::path::PathBuf, Vec<std::path::PathBuf>, Arc<Mutex<VecDeque<TurnEvent>>>, Arc<AtomicBool>) {
(
file_paths.to_vec(),
session_dir.to_path_buf(),
workspaces.to_vec(),
turn_events.clone(),
abort_flag,
)
}
/// Spawn a background subagent that generates tests for modified files.
///
/// Uses the test-generator prompt and has read-write access so it can
/// create test files. Runs in a separate OS thread and reports completion
/// via `TurnEvent::SystemNote { kind: "bg-test-gen" }`.
///
/// Skipped (no-op) if a test-gen run is already in flight (guarded by
/// `TEST_GEN_RUNNING`) — prevents a chatty multi-turn edit session from
/// stacking overlapping runs. `abort_flag` is forwarded to the generic
/// spawner so the run can be cancelled if the turn aborts.
pub fn spawn_background_test_gen(
file_paths: &[String],
session_dir: &Path,
@@ -265,29 +273,15 @@ pub fn spawn_background_test_gen(
turn_events: &Arc<Mutex<VecDeque<TurnEvent>>>,
abort_flag: Arc<AtomicBool>,
) {
let (fps, sd, ws, te, af) = review_args(file_paths, session_dir, workspaces, turn_events, abort_flag);
spawn_background_review(
"bg-test-gen",
&TEST_GEN_RUNNING,
crate::prompts::TEST_GENERATOR_PROMPT,
"test-generator",
"coder",
file_paths.to_vec(),
session_dir.to_path_buf(),
workspaces.to_vec(),
turn_events.clone(),
abort_flag,
"bg-test-gen", &TEST_GEN_RUNNING,
crate::prompts::TEST_GENERATOR_PROMPT, "test-generator", "coder",
fps, sd, ws, te, af,
);
}
/// Spawn a background architecture-review subagent.
///
/// Inspects the modified files for architectural consistency (layering,
/// coupling, module boundaries). Reports via
/// `TurnEvent::SystemNote { kind: "bg-arch-review" }`.
///
/// Skipped (no-op) if an arch-review run is already in flight (guarded by
/// `ARCH_REVIEW_RUNNING`). `abort_flag` is forwarded to the generic
/// spawner so the run can be cancelled if the turn aborts.
pub fn spawn_background_arch_review(
file_paths: &[String],
session_dir: &Path,
@@ -295,31 +289,18 @@ pub fn spawn_background_arch_review(
turn_events: &Arc<Mutex<VecDeque<TurnEvent>>>,
abort_flag: Arc<AtomicBool>,
) {
let (fps, sd, ws, te, af) = review_args(file_paths, session_dir, workspaces, turn_events, abort_flag);
spawn_background_review(
"bg-arch-review",
&ARCH_REVIEW_RUNNING,
crate::prompts::ARCH_REVIEWER_PROMPT,
"arch-reviewer",
"reviewer",
file_paths.to_vec(),
session_dir.to_path_buf(),
workspaces.to_vec(),
turn_events.clone(),
abort_flag,
"bg-arch-review", &ARCH_REVIEW_RUNNING,
crate::prompts::ARCH_REVIEWER_PROMPT, "arch-reviewer", "reviewer",
fps, sd, ws, te, af,
);
}
/// Spawn a background security-review subagent.
///
/// Checks modified files for security vulnerabilities. Reports via
/// `TurnEvent::SystemNote { kind: "bg-security-review" }`.
///
/// Only reviews production code files for security — test files and
/// config files are out of scope for security review.
///
/// Skipped (no-op) if a security-review run is already in flight (guarded by
/// `SECURITY_REVIEW_RUNNING`). `abort_flag` is forwarded to the generic
/// spawner so the run can be cancelled if the turn aborts.
pub fn spawn_background_security_review(
file_paths: &[String],
session_dir: &Path,
@@ -327,25 +308,16 @@ pub fn spawn_background_security_review(
turn_events: &Arc<Mutex<VecDeque<TurnEvent>>>,
abort_flag: Arc<AtomicBool>,
) {
// Only review production code files for security — test files and
// config files are out of scope for security review.
let (_, sd, ws, te, af) = review_args(file_paths, session_dir, workspaces, turn_events, abort_flag);
let prod_paths: Vec<String> = file_paths
.iter()
.filter(|p| is_production_code(p))
.cloned()
.collect();
spawn_background_review(
"bg-security-review",
&SECURITY_REVIEW_RUNNING,
crate::prompts::SECURITY_REVIEWER_PROMPT,
"security-reviewer",
"reviewer",
prod_paths,
session_dir.to_path_buf(),
workspaces.to_vec(),
turn_events.clone(),
abort_flag,
"bg-security-review", &SECURITY_REVIEW_RUNNING,
crate::prompts::SECURITY_REVIEWER_PROMPT, "security-reviewer", "reviewer",
prod_paths, sd, ws, te, af,
);
}
+107 -111
View File
@@ -14,9 +14,28 @@ use super::workspace::generate_workspace_tree;
use crate::dto::chat::message::ChatMessage;
use crate::dto::provider::request::ToolDef;
use crate::tool::tool_is_risky;
use sha2::Digest;
use crate::app::util::backoff::backoff_seconds;
use std::time::Duration;
use tokio::sync::mpsc;
use zesdex_cms::domain::repository::EditLogRepository;
/// Exponential backoff with ±25% jitter for subagent step retries, capped at 16s.
fn step_retry_delay(attempt: u32) -> Duration {
backoff_seconds(attempt, 16)
}
/// Heuristic to decide whether the error is worth retrying.
fn should_retry_subagent_step(err_str: &str) -> bool {
// Never retry auth/billing failures
if crate::service::provider::is_auth_error(err_str) {
return false;
}
// Never retry abort or user cancellation
if err_str.to_lowercase().contains("aborted") {
return false;
}
// Everything else (timeout, 5xx, rate-limit, network blip) is retryable
true
}
fn format_subagent_progress(prefix: &str, text: &str) -> String {
let lines: Vec<&str> = text.lines().filter(|l| !l.trim().is_empty()).collect();
@@ -98,11 +117,7 @@ pub fn run_subagent(
for step in 0..ctx.max_steps {
// Check abort flag before each LLM call so a stuck subagent can
// be cancelled from the parent (mirrors main agent behaviour).
if ctx
.abort_flag
.as_ref()
.is_some_and(|f| f.load(std::sync::atomic::Ordering::SeqCst))
{
if crate::app::util::abort::is_aborted(&ctx.abort_flag) {
let _ = tx.blocking_send(SubagentEvent::StepFailed {
step,
error: "subagent aborted by parent".to_string(),
@@ -118,71 +133,87 @@ pub fn run_subagent(
// Use streaming API so the abort flag is checked per SSE event,
// making the subagent responsive to cancellation even during an
// LLM call (non-streaming would block for 10-30s unchecked).
let stream_result = client.chat_with_tools_streaming(
&messages,
tdefs_opt.clone(),
Some(0.7),
Some(4096),
|event| -> bool {
// Check abort on every SSE event for responsive cancellation.
if ctx
.abort_flag
.as_ref()
.is_some_and(|f| f.load(std::sync::atomic::Ordering::SeqCst))
{
return false; // signals provider to abort
}
match event {
crate::app::runtime::stream::StreamEvent::Reasoning(text) => {
current_thinking.push_str(text);
let prog = format_subagent_progress("thinking", &current_thinking);
let _ = tx_clone.blocking_send(SubagentEvent::Progress(prog));
}
crate::app::runtime::stream::StreamEvent::Token(text) => {
current_token.push_str(text);
let prog = format_subagent_progress("replying", &current_token);
let _ = tx_clone.blocking_send(SubagentEvent::Progress(prog));
}
crate::app::runtime::stream::StreamEvent::Usage {
prompt_tokens,
completion_tokens,
..
} => {
// Capture usage so the drain thread can route it
// to the parent's `UsageStats::review_tokens`.
// Last writer wins — providers send exactly one
// Usage event per streaming call.
step_usage = Some((*prompt_tokens, *completion_tokens));
}
_ => {}
}
true
},
);
// Retry the LLM call at the step level (up to 3 attempts) so a
// transient network blip doesn't kill the subagent. The underlying
// `chat_with_tools_streaming` already has its own retry loop (5 +
// non-streaming fallback), so this loop is a second safety net for
// rare cases where the combined 5+10 retries are all exhausted.
let max_step_retries = 3;
let mut step_attempt = 0u32;
let (response, returned_usage) = match stream_result {
Ok(result) => result,
Err(e) => {
let is_abort = ctx
.abort_flag
.as_ref()
.is_some_and(|f| f.load(std::sync::atomic::Ordering::SeqCst))
|| e.to_string().contains("aborted");
let _ = tx.blocking_send(SubagentEvent::StepFailed {
step,
error: if is_abort {
"subagent aborted by user".to_string()
} else {
e.to_string()
},
});
if is_abort {
anyhow::bail!("subagent aborted by parent at step {step}");
let (response, returned_usage) = loop {
step_attempt += 1;
let stream_result = client.chat_with_tools_streaming(
&messages,
tdefs_opt.clone(),
Some(0.7),
Some(4096),
|event| -> bool {
// Check abort on every SSE event for responsive cancellation.
if crate::app::util::abort::is_aborted(&ctx.abort_flag) {
return false; // signals provider to abort
}
match event {
crate::app::runtime::stream::StreamEvent::Reasoning(text) => {
current_thinking.push_str(text);
let prog = format_subagent_progress("thinking", &current_thinking);
let _ = tx_clone.blocking_send(SubagentEvent::Progress(prog));
}
crate::app::runtime::stream::StreamEvent::Token(text) => {
current_token.push_str(text);
let prog = format_subagent_progress("replying", &current_token);
let _ = tx_clone.blocking_send(SubagentEvent::Progress(prog));
}
crate::app::runtime::stream::StreamEvent::Usage {
prompt_tokens,
completion_tokens,
..
} => {
// Capture usage so the drain thread can route it
// to the parent's `UsageStats::review_tokens`.
// Last writer wins — providers send exactly one
// Usage event per streaming call.
step_usage = Some((*prompt_tokens, *completion_tokens));
}
_ => {}
}
true
},
ctx.abort_flag.as_deref(),
);
match stream_result {
Ok(result) => break result,
Err(e) => {
let err_str = e.to_string();
let is_abort = crate::app::util::abort::is_aborted(&ctx.abort_flag)
|| err_str.contains("aborted");
if is_abort || !should_retry_subagent_step(&err_str) || step_attempt >= max_step_retries {
let _ = tx.blocking_send(SubagentEvent::StepFailed {
step,
error: if is_abort {
"subagent aborted by user".to_string()
} else {
err_str.clone()
},
});
if is_abort {
anyhow::bail!("subagent aborted by parent at step {step}");
}
anyhow::bail!("subagent call failed at step {step} after {step_attempt} attempt(s): {err_str}");
}
let delay = step_retry_delay(step_attempt);
tracing::warn!(
"[subagent] step {step} attempt {step_attempt}/{max_step_retries} failed: {err_str}. \
retrying in {delay:?}...",
);
let _ = tx.blocking_send(SubagentEvent::Progress(format!(
"retrying step {step} ({step_attempt}/{max_step_retries}) after error…",
)));
std::thread::sleep(delay);
}
// No non-streaming fallback — API must support streaming.
// Non-streaming calls block for up to 1 min without checking
// abort_flag, making cancellation unresponsive.
anyhow::bail!("subagent call failed at step {step}: {e}");
}
};
@@ -237,7 +268,7 @@ pub fn run_subagent(
for tool_call in &tool_calls {
let handle = s.spawn(move || {
// Check abort flag before each tool execution
if ctx.abort_flag.as_ref().is_some_and(|f| f.load(std::sync::atomic::Ordering::SeqCst)) {
if crate::app::util::abort::is_aborted(&ctx.abort_flag) {
return (tool_call, Err(anyhow::anyhow!("subagent aborted by parent during tool execution")));
}
@@ -284,49 +315,14 @@ pub fn run_subagent(
let run_res = tool.run(tool_ctx_ref, &args);
if is_edit && run_res.is_ok() {
let reason = args
.get("reason")
.and_then(|v| v.as_str())
.unwrap_or("unnamed");
let path = args
.get("path")
.and_then(|v| v.as_str())
.unwrap_or("unknown");
let content_sha256 = {
let content = args.get("content").or_else(|| args.get("new"));
let hash = sha2::Sha256::digest(
content.and_then(|v| v.as_str()).unwrap_or("").as_bytes(),
);
hex::encode(hash)
};
let bytes_delta = if tool_name == "write" {
args.get("content")
.and_then(|v| v.as_str())
.map_or(0, |s| s.len() as i64)
} else {
let old = args.get("old").and_then(|v| v.as_str()).unwrap_or("");
let new = args.get("new").and_then(|v| v.as_str()).unwrap_or("");
(new.len() as i64 - old.len() as i64).abs()
};
let session_id = ctx.session_dir
.file_name()
.and_then(|n| n.to_str())
.unwrap_or("unknown")
.to_string();
let entry = zesdex_cms::domain::edit_log::EditLogEntry {
ts: chrono::Utc::now().timestamp_millis(),
tool: tool_name.clone(),
path: path.to_string(),
reason: reason.to_string(),
content_sha256,
bytes_delta,
origin: tool_ctx_ref.origin.tag(),
session_id,
};
let repo = zesdex_cms::infrastructure::persistence::edit_log_repo::JsonlEditLogRepository::new();
if let Ok(mut el) = repo.open(&ctx.session_dir) {
let _ = repo.append(&ctx.session_dir, &mut el, entry);
}
.unwrap_or("unknown");
crate::tool::log_write_edit_tool(
&args, tool_name, &tool_ctx_ref.origin.tag(),
&ctx.session_dir, session_id,
);
}
run_res
}
@@ -27,40 +27,19 @@ pub(crate) fn resolve_provider_config() -> (String, String, Option<String>, Stri
.load(&store_base_dir)
.unwrap_or_default();
let mut api_key = settings
.api_keys
.get(&settings.provider)
.cloned()
.unwrap_or_else(|| {
tracing::warn!(
"[subagent] no API key for provider '{}' in settings, trying env/default",
settings.provider
);
String::new()
});
let api_key = crate::service::provider::resolve_api_key(&settings, &app_config);
if api_key.is_empty() {
tracing::warn!(
"[subagent] all API key resolution paths exhausted for '{}'",
settings.provider
);
}
let model = settings.model.clone();
let base_url = app_config
.providers
.get(&settings.provider)
.map(|p| p.api_base.clone());
if api_key.is_empty() {
if let Some(provider_cfg) = app_config.providers.get(&settings.provider) {
api_key = provider_cfg
.api_key_env
.as_ref()
.and_then(|env| std::env::var(env).ok())
.or_else(|| provider_cfg.default_api_key.clone())
.unwrap_or_else(|| {
tracing::warn!(
"[subagent] all API key resolution paths exhausted for '{}'",
settings.provider
);
String::new()
});
}
}
(api_key, model, base_url, settings.provider)
}
@@ -0,0 +1,27 @@
//! Shared abort-flag checks.
//!
//! Three variants (Option<Arc<AtomicBool>>, bare AtomicBool, and
//! Option<&AtomicBool>) cover the agent runtime, subagent, workflow
//! engine, and provider.
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
/// Check whether an optional abort flag has been signalled.
pub fn is_aborted(flag: &Option<Arc<AtomicBool>>) -> bool {
flag.as_ref().is_some_and(|f| f.load(Ordering::SeqCst))
}
/// Check whether a bare abort flag has been signalled.
pub fn is_aborted_direct(flag: &AtomicBool) -> bool {
flag.load(Ordering::SeqCst)
}
/// Check whether an optional borrowed abort flag has been signalled.
///
/// This variant handles the `Option<&AtomicBool>` pattern used in
/// service/provider.rs where the flag is passed as a by-value optional
/// reference rather than an `Arc`.
pub fn is_aborted_ref(flag: Option<&AtomicBool>) -> bool {
flag.is_some_and(|f| f.load(Ordering::SeqCst))
}
@@ -0,0 +1,30 @@
//! Exponential backoff with jitter.
//!
//! Three use cases (subagent, provider, workflow) all share the same formula
//! with different caps. This module provides a single implementation.
use std::time::{Duration, SystemTime, UNIX_EPOCH};
/// Compute an exponential backoff with ±25% jitter.
///
/// `attempt` is 0-based (first retry -> attempt=0 -> base=1s,
/// second retry -> attempt=1 -> base=2s, etc.).
/// `max_secs` sets the cap.
pub fn backoff_seconds(attempt: u32, max_secs: u64) -> Duration {
let base_secs = (2u64).pow(attempt).min(max_secs);
let quarter = (base_secs * 250_000_000).max(100_000_000); // 25% of base, min 100ms
let offset = jitter_ns(quarter * 2); // [0, 50% of base)
// ±25%: offset in [0, 2×quarter), result = base + offset - quarter
// which lies in [base - 25%, base + 25%).
let ns = base_secs * 1_000_000_000 + offset - quarter;
Duration::from_nanos(ns)
}
/// Return a jitter offset in the range [0, range_ns).
fn jitter_ns(range_ns: u64) -> u64 {
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos() as u64;
nanos % range_ns
}
@@ -0,0 +1,4 @@
//! Utility modules for shared helpers.
pub mod abort;
pub mod backoff;
@@ -25,7 +25,7 @@ pub(crate) use primitives::PrimitiveCtx;
use serde::{Deserialize, Serialize};
use std::sync::{
atomic::{AtomicBool, Ordering},
atomic::AtomicBool,
Arc, Mutex,
};
use std::time::Duration;
@@ -376,10 +376,7 @@ fn spawn_single_agent(sp: SpawnCtx<'_>) -> anyhow::Result<String> {
});
// Check abort before even starting the subagent.
if sp
.abort_flag
.as_ref()
.is_some_and(|f| f.load(Ordering::SeqCst))
if crate::app::util::abort::is_aborted(sp.abort_flag)
{
anyhow::bail!("subagent '{}' aborted before start", sp.agent_name);
}
@@ -387,13 +384,54 @@ fn spawn_single_agent(sp: SpawnCtx<'_>) -> anyhow::Result<String> {
// Run subagent on a separate thread so the abort flag can be polled.
// If abort is requested while the subagent is running, we abandon the
// thread (Rust threads cannot be forcibly killed) and return early.
//
// Retry: wrap `run_subagent` with up to 2 attempts so a transient
// network blip doesn't kill the whole pipeline. Auth and abort errors
// are not retried.
let (done_tx, done_rx) = std::sync::mpsc::channel::<anyhow::Result<String>>();
let bg_ctx = ctx;
let bg_tx = tx;
let bg_name = sp.agent_name.to_string();
let bg_abort = sp.abort_flag.clone();
let bg_abort_thread = bg_abort.clone();
let bg_name_thread = bg_name.clone();
std::thread::spawn(move || {
let _ = done_tx.send(run_subagent(&bg_ctx, &bg_tx));
// Retry wrapper: jittered backoff with 8s cap.
let retry_backoff =
|attempt: u32| crate::app::util::backoff::backoff_seconds(attempt, 8);
for attempt in 1..=2 {
// Don't retry if aborted.
if crate::app::util::abort::is_aborted(&bg_abort_thread)
{
let _ = done_tx.send(Err(anyhow::anyhow!(
"subagent '{bg_name_thread}' aborted by user"
)));
return;
}
match run_subagent(&bg_ctx, &bg_tx) {
Ok(output) => {
let _ = done_tx.send(Ok(output));
return;
}
Err(e) => {
let err_str = e.to_string();
let is_auth = crate::service::provider::is_auth_error(&err_str);
// Auth errors are permanent — don't retry.
if is_auth || attempt >= 2 {
let _ = done_tx.send(Err(e));
return;
}
tracing::warn!(
"[workflow] agent '{bg_name_thread}' attempt {attempt}/2 failed: {err_str}. retrying...",
);
std::thread::sleep(retry_backoff(attempt));
}
}
}
// Should be unreachable because the loop returns on success or final
// error, but keep the compiler happy.
unreachable!()
});
let poll_interval = Duration::from_millis(200);
@@ -410,7 +448,7 @@ fn spawn_single_agent(sp: SpawnCtx<'_>) -> anyhow::Result<String> {
"subagent '{bg_name}' timed out after {timeout}ms",
));
}
if bg_abort.as_ref().is_some_and(|f| f.load(Ordering::SeqCst)) {
if crate::app::util::abort::is_aborted(&bg_abort) {
break Err(anyhow::anyhow!("subagent '{bg_name}' aborted by user"));
}
}
@@ -419,7 +457,7 @@ fn spawn_single_agent(sp: SpawnCtx<'_>) -> anyhow::Result<String> {
if let Ok(r) = done_rx.recv_timeout(poll_interval) {
break r;
}
if bg_abort.as_ref().is_some_and(|f| f.load(Ordering::SeqCst)) {
if crate::app::util::abort::is_aborted(&bg_abort) {
break Err(anyhow::anyhow!("subagent '{bg_name}' aborted by user"));
}
}
+42 -86
View File
@@ -10,7 +10,12 @@ use crate::app::state::input::AutocompleteKind;
use crate::app::state::rest::AppStateRest;
use crate::app::state::types::Overlay;
use crate::controller::command::parse_command;
use zesdex_cms::domain::repository::SettingsRepository;
/// Mark state dirty and return an empty action list.
fn mark(state: &mut AppStateRest) -> Vec<Action> {
state.mark_dirty();
Vec::new()
}
/// Translate a terminal `KeyEvent` into zero or more `Action` values
/// based on the current application state.
@@ -33,17 +38,11 @@ pub fn handle_key(key: KeyEvent, state: &mut AppStateRest) -> Vec<Action> {
if let Some(ref ed) = state.misc.editor.clone() {
let content = ed.as_string();
if let Err(e) = std::fs::write(&ed.path, &content) {
state.push_toast(crate::app::state::types::Toast::new(
crate::app::state::types::ToastKind::Error,
format!("Save failed: {e}"),
));
state.toast_error(format!("Save failed: {e}"));
} else {
state.push_toast(crate::app::state::types::Toast::new(
crate::app::state::types::ToastKind::Success,
format!("Saved {}", ed.path),
));
state.toast_success(format!("Saved {}", ed.path));
}
state.dirty = true;
state.mark_dirty();
}
return vec![];
}
@@ -54,7 +53,7 @@ pub fn handle_key(key: KeyEvent, state: &mut AppStateRest) -> Vec<Action> {
KeyCode::Backspace => {
if let Some(ref mut ed) = state.misc.editor {
ed.delete_left();
state.dirty = true;
state.mark_dirty();
}
return vec![];
}
@@ -81,24 +80,14 @@ pub fn handle_key(key: KeyEvent, state: &mut AppStateRest) -> Vec<Action> {
KeyCode::Up => {
let items = crate::app::mode::learning::get_learning_items(state);
let n = items.len();
state.misc.selected_index = if state.misc.selected_index == 0 {
n.saturating_sub(1)
} else {
state.misc.selected_index - 1
};
state.dirty = true;
return vec![];
state.misc.selected_index = crate::app::mode::cycle_selected_index(state.misc.selected_index, n, false);
return mark(state);
}
KeyCode::Down => {
let items = crate::app::mode::learning::get_learning_items(state);
let n = items.len();
state.misc.selected_index = if n == 0 {
0
} else {
(state.misc.selected_index + 1) % n
};
state.dirty = true;
return vec![];
state.misc.selected_index = crate::app::mode::cycle_selected_index(state.misc.selected_index, n, true);
return mark(state);
}
KeyCode::Enter | KeyCode::Char('a') => {
let items = crate::app::mode::learning::get_learning_items(state);
@@ -155,10 +144,7 @@ pub fn handle_key(key: KeyEvent, state: &mut AppStateRest) -> Vec<Action> {
state.misc.pending_clipboard_copy = Some(msg.content.clone());
}
None => {
state.push_toast(crate::app::state::types::Toast::new(
crate::app::state::types::ToastKind::Info,
"No assistant message to copy yet".to_string(),
));
state.toast_info("No assistant message to copy yet".to_string());
}
}
Vec::new()
@@ -166,8 +152,7 @@ pub fn handle_key(key: KeyEvent, state: &mut AppStateRest) -> Vec<Action> {
KeyCode::Enter => {
if state.input.autocomplete_visible {
state.input.select_autocomplete();
state.dirty = true;
return Vec::new();
return mark(state);
}
if state.misc.overlay.is_active() {
return handle_overlay_enter(state);
@@ -181,16 +166,14 @@ pub fn handle_key(key: KeyEvent, state: &mut AppStateRest) -> Vec<Action> {
KeyCode::Backspace => {
if state.input.autocomplete_visible {
state.input.close_autocomplete();
state.dirty = true;
return Vec::new();
return mark(state);
}
vec![Action::DeleteChar]
}
KeyCode::Delete => {
if state.input.autocomplete_visible {
state.input.close_autocomplete();
state.dirty = true;
return Vec::new();
return mark(state);
}
vec![Action::DeleteCharRight]
}
@@ -203,28 +186,20 @@ pub fn handle_key(key: KeyEvent, state: &mut AppStateRest) -> Vec<Action> {
KeyCode::Up => {
if state.input.autocomplete_visible {
state.input.cycle_autocomplete(false);
state.dirty = true;
state.mark_dirty();
Vec::new()
} else if state.misc.overlay == Overlay::Effort {
mode::effort::cycle_effort(state);
Vec::new()
} else if state.misc.overlay == Overlay::Rewind {
let n = mode::rewind::rewind_count(state);
state.misc.selected_index = if state.misc.selected_index == 0 {
n.saturating_sub(1)
} else {
state.misc.selected_index - 1
};
state.dirty = true;
state.misc.selected_index = mode::cycle_selected_index(state.misc.selected_index, n, false);
state.mark_dirty();
Vec::new()
} else if state.misc.overlay == Overlay::ModelSelector {
let n = state.app_config.providers.len();
state.misc.selected_index = if state.misc.selected_index == 0 {
n.saturating_sub(1)
} else {
state.misc.selected_index - 1
};
state.dirty = true;
state.misc.selected_index = mode::cycle_selected_index(state.misc.selected_index, n, false);
state.mark_dirty();
Vec::new()
} else if key.modifiers.contains(KeyModifiers::CONTROL) {
vec![Action::ScrollUp]
@@ -235,28 +210,20 @@ pub fn handle_key(key: KeyEvent, state: &mut AppStateRest) -> Vec<Action> {
KeyCode::Down => {
if state.input.autocomplete_visible {
state.input.cycle_autocomplete(true);
state.dirty = true;
state.mark_dirty();
Vec::new()
} else if state.misc.overlay == Overlay::Effort {
mode::effort::cycle_effort(state);
Vec::new()
} else if state.misc.overlay == Overlay::Rewind {
let n = mode::rewind::rewind_count(state);
state.misc.selected_index = if n == 0 {
0
} else {
(state.misc.selected_index + 1) % n
};
state.dirty = true;
state.misc.selected_index = mode::cycle_selected_index(state.misc.selected_index, n, true);
state.mark_dirty();
Vec::new()
} else if state.misc.overlay == Overlay::ModelSelector {
let n = state.app_config.providers.len();
state.misc.selected_index = if n == 0 {
0
} else {
(state.misc.selected_index + 1) % n
};
state.dirty = true;
state.misc.selected_index = mode::cycle_selected_index(state.misc.selected_index, n, true);
state.mark_dirty();
Vec::new()
} else if key.modifiers.contains(KeyModifiers::CONTROL) {
vec![Action::ScrollDown]
@@ -275,7 +242,7 @@ pub fn handle_key(key: KeyEvent, state: &mut AppStateRest) -> Vec<Action> {
vec![Action::AbortTurn]
} else if state.input.autocomplete_visible {
state.input.close_autocomplete();
state.dirty = true;
state.mark_dirty();
Vec::new()
} else if state.misc.overlay.is_active() {
vec![Action::CloseOverlay]
@@ -290,24 +257,24 @@ pub fn handle_key(key: KeyEvent, state: &mut AppStateRest) -> Vec<Action> {
} else {
state.input.tab_complete();
}
state.dirty = true;
state.mark_dirty();
} else if state.input.autocomplete_kind == AutocompleteKind::FileMention
&& state.input.autocomplete_visible
{
state.input.cycle_autocomplete(true);
state.dirty = true;
state.mark_dirty();
}
Vec::new()
}
KeyCode::Char(c) => {
if state.input.autocomplete_visible {
state.input.close_autocomplete();
state.dirty = true;
state.mark_dirty();
}
// Insert the character inline so we can immediately check the
// new buffer state for autocomplete triggers.
state.input.insert(c);
state.dirty = true;
state.mark_dirty();
// Show autocomplete immediately when the buffer starts with `/`,
// without requiring an extra Tab press.
if state.input.buffer.starts_with('/') {
@@ -338,7 +305,7 @@ fn handle_overlay_enter(state: &mut AppStateRest) -> Vec<Action> {
}
Overlay::Settings => {
mode::settings::cycle_internet_mode(&mut state.settings);
state.dirty = true;
state.mark_dirty();
Vec::new()
}
Overlay::Todo => {
@@ -359,16 +326,12 @@ fn handle_overlay_enter(state: &mut AppStateRest) -> Vec<Action> {
.api_keys
.insert(state.settings.provider.clone(), text.clone());
}
let _ = zesdex_cms::infrastructure::persistence::settings_repo::JsonSettingsRepository::new()
.save(&state.store_base_dir(), &state.settings);
state.save_settings();
state.input.buffer.clear();
state.input.cursor = 0;
state.misc.overlay = Overlay::None;
state.push_toast(crate::app::state::types::Toast::new(
crate::app::state::types::ToastKind::Success,
"API key saved".to_string(),
));
state.dirty = true;
state.toast_success("API key saved".to_string());
state.mark_dirty();
Vec::new()
}
@@ -406,25 +369,18 @@ fn handle_overlay_enter(state: &mut AppStateRest) -> Vec<Action> {
{
state.settings.api_keys.insert(provider.clone(), env_key);
}
let _ = zesdex_cms::infrastructure::persistence::settings_repo::JsonSettingsRepository::new()
.save(&state.store_base_dir(), &state.settings);
state.push_toast(crate::app::state::types::Toast::new(
crate::app::state::types::ToastKind::Success,
format!("Switched to {provider} / {model}"),
));
state.save_settings();
state.toast_success(format!("Switched to {provider} / {model}"));
}
}
state.misc.overlay = Overlay::None;
state.dirty = true;
state.mark_dirty();
Vec::new()
}
Overlay::ClearConfirm => {
state.push_toast(crate::app::state::types::Toast::new(
crate::app::state::types::ToastKind::Info,
"Transcript cleared".to_string(),
));
state.toast_info("Transcript cleared".to_string());
state.misc.overlay = Overlay::None;
state.dirty = true;
state.mark_dirty();
Vec::new()
}
+224 -36
View File
@@ -1,9 +1,35 @@
//! Blocking HTTP client for OpenAI/Anthropic-compatible chat completion APIs,
//! supporting both non-streaming and SSE-streaming requests with automatic retry.
//!
//! # Retry policy
//!
//! Both paths use exponential backoff with ±25% jitter so retries spread out
//! naturally instead of hammering the server in lockstep. Auth errors
//! (401/402/403) are never retried — they indicate a bad key or billing issue
//! that retrying won't fix. Rate-limit (429) responses get a longer backoff
//! (base 5s instead of the usual 1s) so the server has time to drain its queue.
//!
//! ## Non-streaming (`chat_with_tools_non_streaming`)
//! - Up to **10** attempts
//! - Backoff: `1s, 2s, 4s, 8s, 16s, 30s(capped), 30s, …` + jitter
//! - Auth errors → abort immediately on the **status code** embedded in the
//! error message (avoids false positives from port numbers, model names etc.)
//!
//! ## Streaming (`chat_with_tools_streaming`)
//! - Up to **5** attempts *before* any meaningful content (tokens / reasoning)
//! - After meaningful content arrives, falls back to a **non-streaming retry**
//! (the non-streaming call carries 10 retries of its own), so a mid-stream
//! network blip is recovered instead of killing the whole turn.
//! - The `started` flag still prevents retries on the raw SSE call once the
//! stream has begun (partial content cannot be safely replayed), but the
//! caller-level fallback handles that case.
use anyhow::Result;
use std::sync::atomic::AtomicBool;
use std::time::Duration;
use crate::app::runtime::stream::turn::StreamedTurn;
use crate::app::util::backoff::backoff_seconds;
use crate::app::runtime::stream::{SseParser, StreamEvent};
use crate::dto::chat::message::ChatMessage;
use crate::dto::provider::request::{ChatRequest, StreamOptions, ToolDef};
@@ -14,6 +40,54 @@ pub const DEFAULT_API_KEY: &str = "";
const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
const REQUEST_TIMEOUT: Duration = Duration::from_mins(1);
// ---------------------------------------------------------------------------
// Retry helpers
// ---------------------------------------------------------------------------
/// Exponential backoff with ±25% jitter, capped at 30 seconds.
///
/// `attempt` is 1-based (first retry → attempt=1).
fn backoff_duration(attempt: u32) -> Duration {
backoff_seconds(attempt, 30)
}
/// Is the error an auth / billing failure that retrying won't fix?
///
/// Matches the structured "API error {status} from …" format used by the
/// request builders below, plus well-known auth keywords in case the body
/// contains them. This is intentionally tighter than `contains("401")`,
/// which could false-positive on a URL port, model name, or body text.
pub fn is_auth_error(err_str: &str) -> bool {
let err_lower = err_str.to_lowercase();
// Structured HTTP status patterns
(err_str.contains("API error 401")
|| err_str.contains("API error 402")
|| err_str.contains("API error 403"))
// Keyword fallback for non-standard error formats
|| err_lower.contains("unauthorized")
|| err_lower.contains("forbidden")
|| err_lower.contains("authentication failed")
}
/// Is the error a rate-limit response?
fn is_rate_limit(err_str: &str) -> bool {
err_str.contains("API error 429") || err_str.to_lowercase().contains("rate limit")
}
/// Return a rate-appropriate backoff (longer for 429).
fn backoff_for_error(attempt: u32, err_str: &str) -> Duration {
if is_rate_limit(err_str) {
// Rate limits need more time to drain — backoff capped at 60s.
backoff_seconds(attempt, 60)
} else {
backoff_duration(attempt)
}
}
// ---------------------------------------------------------------------------
// Client
// ---------------------------------------------------------------------------
/// Blocking HTTP client for a single LLM provider endpoint.
///
/// Holds the reqwest client, credentials, and model/base URL selection used
@@ -84,11 +158,13 @@ impl LlmClient {
/// Send a non-streaming chat completion request and return the assistant's reply.
///
/// Flow: build request → POST with retry loop (up to 10 attempts, 2s backoff)
/// → parse JSON response → extract first choice's message and token usage.
/// Flow: build request → POST with retry loop (up to 10 attempts, exponential
/// backoff with jitter) → parse JSON response → extract first choice's message
/// and token usage.
///
/// Why: retries transient failures but aborts immediately on 401/403, since
/// those indicate a bad API key that retrying won't fix.
/// Why: retries transient failures but aborts immediately on 401/402/403 (bad
/// API key / billing issue — retrying won't fix). 429 (rate-limit) responses
/// get a longer backoff so the server has time to recover.
///
/// Return: `Err` if all retries are exhausted, an auth error occurs, or the
/// response has no choices.
@@ -96,12 +172,15 @@ impl LlmClient {
&self,
messages: &[ChatMessage],
tools: Option<Vec<ToolDef>>,
max_tokens: Option<u32>,
temperature: Option<f32>,
abort_flag: Option<&AtomicBool>,
) -> Result<(ChatMessage, Option<(u64, u64)>)> {
let req = ChatRequest {
model: self.model.clone(),
messages: messages.to_vec(),
max_tokens: Some(4096),
temperature: Some(0.7),
max_tokens: Some(max_tokens.unwrap_or(4096)),
temperature: Some(temperature.unwrap_or(0.7)),
tools,
stream: Some(false),
stop: None,
@@ -112,11 +191,17 @@ impl LlmClient {
let url = format!("{}/chat/completions", self.base_url);
let max_retries = 10;
let mut attempt = 0;
let mut attempt = 0u32;
loop {
attempt += 1;
// Check abort before each retry so user cancellation is
// responsive even during a long non-streaming backoff chain.
if crate::app::util::abort::is_aborted_ref(abort_flag) {
anyhow::bail!("aborted");
}
let mut http_req = self
.client
.post(&url)
@@ -160,28 +245,45 @@ impl LlmClient {
Ok((msg, usage)) => return Ok((msg, usage)),
Err(e) => {
let err_str = e.to_string();
let err_lower = err_str.to_lowercase();
let is_auth_error = err_str.contains("401")
|| err_str.contains("403")
|| err_lower.contains("unauthorized")
|| err_lower.contains("forbidden")
|| err_lower.contains("authentication failed");
if attempt >= max_retries || is_auth_error {
if attempt >= max_retries || is_auth_error(&err_str) {
return Err(e);
}
tracing::warn!("Warning: {}. Retrying {}/{}...", e, attempt, max_retries);
std::thread::sleep(Duration::from_secs(2));
let delay = backoff_for_error(attempt, &err_str);
tracing::warn!(
"Warning: {}. Retrying {}/{}, sleeping {delay:?}...",
e,
attempt,
max_retries,
);
std::thread::sleep(delay);
}
}
}
}
/// Streaming variant of `chat_with_tools`. Feeds SSE chunks into an `SseParser` /
/// `StreamedTurn` and invokes `on_event` for every parsed `StreamEvent` as it arrives,
/// so the caller can push incremental UI updates in real time. Returns the fully
/// assembled assistant message plus token usage (prompt, completion) if the server
/// reported it. Retries the whole request only if no event has been observed yet
/// (once tokens start arriving, a partial turn cannot be safely replayed).
/// Streaming variant of `chat_with_tools`. Feeds SSE chunks into an
/// `SseParser` / `StreamedTurn` and invokes `on_event` for every parsed
/// `StreamEvent` as it arrives, so the caller can push incremental UI
/// updates in real time.
///
/// Returns the fully assembled assistant message plus token usage (prompt,
/// completion) if the server reported it.
///
/// # Retry semantics
///
/// Retries the raw SSE request only *before* any meaningful content (text
/// tokens or reasoning tokens) has been received — once the LLM has started
/// generating, a partial stream cannot be safely replayed without duplicating
/// or garbling output.
///
/// Once meaningful content has arrived and the stream fails, **this method
/// falls back to a non-streaming call** (which carries its own 10-retry
/// loop). The non-streaming call uses the same `messages` independently
/// (no SSE state to replay), so the caller always gets a complete result if
/// the provider is reachable.
///
/// Auth errors (401/402/403) are never retried on either path. Rate-limit
/// (429) responses get a longer backoff.
pub fn chat_with_tools_streaming(
&self,
messages: &[ChatMessage],
@@ -189,7 +291,11 @@ impl LlmClient {
temperature: Option<f32>,
max_tokens: Option<u32>,
mut on_event: impl FnMut(&StreamEvent) -> bool,
abort_flag: Option<&AtomicBool>,
) -> Result<(ChatMessage, Option<(u64, u64)>)> {
// Clone tools for the non-streaming fallback path — the original
// is moved into the ChatRequest below and cannot be used again.
let tools_for_fallback = tools.clone();
let req = ChatRequest {
model: self.model.clone(),
messages: messages.to_vec(),
@@ -206,37 +312,88 @@ impl LlmClient {
};
let url = format!("{}/chat/completions", self.base_url);
// Fewer retries on streaming because `run_agent_turn` has a
// non-streaming fallback that also retries. Combined total is
// capped implicitly by the per-turn timeout and step limits.
let max_retries = 3;
let mut attempt = 0;
// Phase 1: Retry the raw SSE call up to 5 times, but only before
// meaningful content arrives. After that, fall back to non-streaming.
let max_retries_stream = 5;
let mut attempt = 0u32;
let mut started = false;
// Track whether we've emitted text/reasoning tokens (meaningful
// content). Non-meaningful events (role/usage/done) are safe to
// ignore for the retry decision.
let mut meaningful_content = false;
loop {
attempt += 1;
let mut captured_content = false;
let mut wrapped = |event: &StreamEvent| -> bool {
started = true;
match event {
StreamEvent::Token(_) | StreamEvent::Reasoning(_) => {
captured_content = true;
}
_ => {}
}
on_event(event)
};
match self.try_stream_once(&req, &url, &mut wrapped) {
Ok(result) => return Ok(result),
Err(e) => {
let err_str = e.to_string();
let err_lower = err_str.to_lowercase();
let is_auth_error = err_str.contains("401")
|| err_str.contains("403")
|| err_lower.contains("unauthorized")
|| err_lower.contains("forbidden")
|| err_lower.contains("authentication failed");
if started || attempt >= max_retries || is_auth_error {
if is_auth_error(&err_str) {
return Err(e);
}
tracing::warn!("Warning: {}. Retrying {}/{}...", e, attempt, max_retries);
std::thread::sleep(Duration::from_secs(2));
// Once meaningful content has been streamed, a raw SSE
// retry would produce a different sequence — fall back
// to non-streaming so the caller gets a clean,
// reproducible answer.
if captured_content || started && (attempt >= max_retries_stream) {
meaningful_content = captured_content || meaningful_content;
break;
}
if attempt >= max_retries_stream {
return Err(e);
}
let delay = backoff_for_error(attempt, &err_str);
tracing::warn!(
"Warning: {}. Retrying stream {}/{}, sleeping {delay:?}...",
e,
attempt,
max_retries_stream,
);
std::thread::sleep(delay);
}
}
}
// Phase 2: If we got meaningful content via SSE but the stream
// failed before completion, fall back to a non-streaming retry.
// This preserves the conversation state because the messages
// passed in are the same — we don't need the partial SSE output.
if meaningful_content {
// Check abort before entering the blocking non-streaming
// call — otherwise the fallback ignores user cancellation.
if crate::app::util::abort::is_aborted_ref(abort_flag) {
return Err(anyhow::anyhow!("aborted"));
}
tracing::warn!(
"streaming failed after meaningful content — falling back to non-streaming call",
);
// Use the same messages and tools so the fallback produces
// a response compatible with what the streaming request
// would have returned (including tool definitions).
return self.chat_with_tools_non_streaming(
messages,
tools_for_fallback,
max_tokens,
temperature,
abort_flag,
);
}
Err(anyhow::anyhow!(
"streaming request failed after {max_retries_stream} attempts"
))
}
/// Perform one streaming chat completion request, parsing SSE events until completion.
@@ -347,3 +504,34 @@ impl LlmClient {
Ok((turn.build_assistant_message(), usage))
}
}
/// Resolve the API key for the currently configured provider, falling back
/// through settings → env var → provider default.
///
/// Used by both the main agent turn loop (`spawn.rs`) and subagent provider
/// resolution (`subagent/provider.rs`) to share the identical fallback chain.
///
/// Flow: try `settings.api_keys[provider]` → try `api_key_env` env var →
/// try `default_api_key` from config → return empty string if all paths
/// exhausted (callers must check and reject the empty case).
pub fn resolve_api_key(
settings: &zesdex_cms::domain::settings::Settings,
app_config: &zesdex_cms::domain::app_config::AppConfig,
) -> String {
let mut api_key = settings
.api_keys
.get(&settings.provider)
.cloned()
.unwrap_or_default();
if api_key.is_empty() {
if let Some(provider_cfg) = app_config.providers.get(&settings.provider) {
api_key = provider_cfg
.api_key_env
.as_ref()
.and_then(|env| std::env::var(env).ok())
.or_else(|| provider_cfg.default_api_key.clone())
.unwrap_or_default();
}
}
api_key
}
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Tool: `edit` — replace a substring in a file with a new string.
use super::super::check_graduated_checks;
use super::super::resolve_path;
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Tool: `read` — display file contents with line numbers.
use super::super::resolve_path;
use super::super::Tool;
@@ -1,5 +1,5 @@
use anyhow::Result;
use serde_json::{json, Value};
use serde_json::Value;
use std::fmt::Write;
use crate::tool::{Tool, ToolCtx};
@@ -7,47 +7,19 @@ use crate::tool::{Tool, ToolCtx};
pub struct LspCompletion;
impl Tool for LspCompletion {
fn name(&self) -> &'static str {
"lsp_completion"
}
fn name(&self) -> &'static str { "lsp_completion" }
fn description(&self) -> &'static str {
"Get code completion suggestions at a cursor position from an LSP server. \
`server` is optional if omitted, the server is auto-detected from the file's extension."
}
fn parameters(&self) -> Value {
json!({
"type": "object",
"properties": {
"server": {
"type": "string",
"description": "Name of the connected LSP server. Optional — auto-detected from the file extension if omitted."
},
"path": {
"type": "string",
"description": "Path to the file (relative to workspace root)"
},
"line": {
"type": "integer",
"description": "Line number (0-based)"
},
"column": {
"type": "integer",
"description": "Column number (0-based)"
}
},
"required": ["path", "line", "column"]
})
}
fn parameters(&self) -> Value { super::lsp_cursor_params(false) }
fn run(&self, ctx: &ToolCtx, args: &Value) -> Result<String> {
let result = super::run_lsp_query(ctx, args, |client, uri, line, column| {
let (completion_result, line, column) = super::run_lsp_query(ctx, args, None, |client, uri, line, column| {
client.completion(uri, line, column)
});
match result {
Ok((completion_result, line, column)) => {
})?;
let items = if let Some(items) = completion_result.as_array() {
items.clone()
} else if let Some(arr) =
@@ -114,8 +86,5 @@ impl Tool for LspCompletion {
writeln!(output, " ... and {} more", items.len() - 50).unwrap();
}
Ok(output)
}
Err(e) => Err(e),
}
}
}
@@ -1,5 +1,5 @@
use anyhow::Result;
use serde_json::{json, Value};
use serde_json::Value;
use std::fmt::Write;
use crate::tool::{Tool, ToolCtx};
@@ -7,82 +7,52 @@ use crate::tool::{Tool, ToolCtx};
pub struct LspDefinition;
impl Tool for LspDefinition {
fn name(&self) -> &'static str {
"lsp_definition"
}
fn name(&self) -> &'static str { "lsp_definition" }
fn description(&self) -> &'static str {
"Go to definition: find the location where a symbol is defined. \
`server` is optional if omitted, the server is auto-detected from the file's extension."
}
fn parameters(&self) -> Value {
json!({
"type": "object",
"properties": {
"server": {
"type": "string",
"description": "Name of the connected LSP server. Optional — auto-detected from the file extension if omitted."
},
"path": {
"type": "string",
"description": "Path to the file (relative to workspace root)"
},
"line": {
"type": "integer",
"description": "Line number (0-based)"
},
"column": {
"type": "integer",
"description": "Column number (0-based)"
}
},
"required": ["path", "line", "column"]
})
}
fn parameters(&self) -> Value { super::lsp_cursor_params(false) }
fn run(&self, ctx: &ToolCtx, args: &Value) -> Result<String> {
let result = super::run_lsp_query(ctx, args, |client, uri, line, column| {
let (def_result, _line, _column) = super::run_lsp_query(ctx, args, None, |client, uri, line, column| {
client.goto_definition(uri, line, column)
});
})?;
match result {
Ok((def_result, _line, _column)) => {
if def_result == Value::Null {
return Ok("No definition found at this position.".to_string());
}
let locations = if let Some(loc) = def_result.as_array() {
loc.clone()
} else {
vec![def_result.clone()]
};
if locations.is_empty() {
return Ok("No definition found.".to_string());
}
let mut output = String::from("Definition(s):\n");
for (i, loc) in locations.iter().enumerate().take(10) {
let target_uri = loc.get("uri").and_then(|u| u.as_str()).unwrap_or("?");
let target_range = loc.get("range").or_else(|| loc.get("targetRange"));
let target_start = target_range.and_then(|r| r.get("start"));
let tl = target_start
.and_then(|s| s.get("line"))
.and_then(serde_json::Value::as_i64)
.unwrap_or(0);
let tc = target_start
.and_then(|s| s.get("character"))
.and_then(serde_json::Value::as_i64)
.unwrap_or(0);
let path_str = target_uri.strip_prefix("file://").unwrap_or(target_uri);
writeln!(output, " {}. {}:{}:{}", i + 1, path_str, tl + 1, tc + 1).unwrap();
}
if locations.len() > 10 {
writeln!(output, " ... and {} more", locations.len() - 10).unwrap();
}
Ok(output)
}
Err(e) => Err(e),
if def_result == Value::Null {
return Ok("No definition found at this position.".to_string());
}
let locations = if let Some(loc) = def_result.as_array() {
loc.clone()
} else {
vec![def_result.clone()]
};
if locations.is_empty() {
return Ok("No definition found.".to_string());
}
let mut output = String::from("Definition(s):\n");
for (i, loc) in locations.iter().enumerate().take(10) {
let target_uri = loc.get("uri").and_then(|u| u.as_str()).unwrap_or("?");
let target_range = loc.get("range").or_else(|| loc.get("targetRange"));
let target_start = target_range.and_then(|r| r.get("start"));
let tl = target_start
.and_then(|s| s.get("line"))
.and_then(serde_json::Value::as_i64)
.unwrap_or(0);
let tc = target_start
.and_then(|s| s.get("character"))
.and_then(serde_json::Value::as_i64)
.unwrap_or(0);
let path_str = target_uri.strip_prefix("file://").unwrap_or(target_uri);
writeln!(output, " {}. {}:{}:{}", i + 1, path_str, tl + 1, tc + 1).unwrap();
}
if locations.len() > 10 {
writeln!(output, " ... and {} more", locations.len() - 10).unwrap();
}
Ok(output)
}
}
+5 -40
View File
@@ -1,5 +1,5 @@
use anyhow::Result;
use serde_json::{json, Value};
use serde_json::Value;
use std::fmt::Write;
use crate::tool::{Tool, ToolCtx};
@@ -7,51 +7,19 @@ use crate::tool::{Tool, ToolCtx};
pub struct LspHover;
impl Tool for LspHover {
fn name(&self) -> &'static str {
"lsp_hover"
}
fn name(&self) -> &'static str { "lsp_hover" }
fn description(&self) -> &'static str {
"Get hover information (type signature, documentation) at a cursor position in a file. \
`server` is optional if omitted, the server is auto-detected from the file's extension."
}
fn parameters(&self) -> Value {
json!({
"type": "object",
"properties": {
"server": {
"type": "string",
"description": "Name of the connected LSP server. Optional — auto-detected from the file extension if omitted."
},
"path": {
"type": "string",
"description": "Path to the file (relative to workspace root)"
},
"line": {
"type": "integer",
"description": "Line number (0-based)"
},
"column": {
"type": "integer",
"description": "Column number (0-based)"
},
"language_id": {
"type": "string",
"description": "Language identifier (e.g. 'rust', 'typescript'). Optional if already set via lsp_connect."
}
},
"required": ["path", "line", "column"]
})
}
fn parameters(&self) -> Value { super::lsp_cursor_params(true) }
fn run(&self, ctx: &ToolCtx, args: &Value) -> Result<String> {
let result = super::run_lsp_query(ctx, args, |client, uri, line, column| {
let (hover_result, _line, _column) = super::run_lsp_query(ctx, args, None, |client, uri, line, column| {
client.hover(uri, line, column)
});
match result {
Ok((hover_result, _line, _column)) => {
})?;
if hover_result == Value::Null {
return Ok("No hover information available at this position.".to_string());
}
@@ -78,9 +46,6 @@ impl Tool for LspHover {
.push_str(&serde_json::to_string_pretty(&hover_result).unwrap_or_default());
}
Ok(output)
}
Err(e) => Err(e),
}
}
}
+57 -10
View File
@@ -1,10 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
mod connect;
mod completion;
mod definition;
@@ -50,6 +43,48 @@ fn known_extensions_for(language_id: &str) -> &[&'static str] {
}
}
/// Build the standard `server` + `path` + `line` + `column` parameter schema
/// used by cursor-based LSP tools (definition, references, completion).
///
/// When `with_language_id` is `true`, an optional `language_id` property is
/// included (for tools like hover that pass it to `didOpen`).
pub fn lsp_cursor_params(with_language_id: bool) -> serde_json::Value {
let mut props = serde_json::json!({
"server": {
"type": "string",
"description": "Name of the connected LSP server. Optional — auto-detected from the file extension if omitted."
},
"path": {
"type": "string",
"description": "Path to the file (relative to workspace root)"
},
"line": {
"type": "integer",
"description": "Line number (0-based)"
},
"column": {
"type": "integer",
"description": "Column number (0-based)"
}
});
if with_language_id {
if let Some(obj) = props.as_object_mut() {
obj.insert(
"language_id".to_string(),
serde_json::json!({
"type": "string",
"description": "Language identifier (e.g. 'rust', 'typescript'). Optional if already set via lsp_connect."
}),
);
}
}
serde_json::json!({
"type": "object",
"properties": props,
"required": ["path", "line", "column"]
})
}
/// Guess which connected LSP server should handle `path` based on its extension.
///
/// Flow: extract extension from `path` -> for each connected server, check
@@ -132,7 +167,16 @@ fn resolve_server_name(ctx: &ToolCtx, args: &Value, path: &str) -> Result<String
/// Opens the file on the server via `didOpen`, invokes the query closure,
/// then closes the file via `didClose`. Returns the query result along with
/// the 0-based line and column for post-processing.
fn run_lsp_query<F, R>(ctx: &ToolCtx, args: &Value, op: F) -> Result<(R, u32, u32)>
///
/// When `text` is `Some`, the provided content is used instead of reading
/// from disk (used by `LspDiagnostics` which receives the full text as an
/// argument).
fn run_lsp_query<F, R>(
ctx: &ToolCtx,
args: &Value,
text: Option<&str>,
op: F,
) -> Result<(R, u32, u32)>
where
F: FnOnce(&mut crate::app::lsp::LspClient, &str, u32, u32) -> Result<R>,
{
@@ -150,8 +194,11 @@ where
let abs_path = crate::tool::resolve_path(&ctx.workspaces, &rel_path)?;
let uri = path_to_lsp_uri(&abs_path.to_string_lossy());
let file_content =
std::fs::read_to_string(&abs_path).map_err(|e| anyhow!("failed to read file '{rel_path}': {e}"))?;
let file_content = match text {
Some(t) => t.to_string(),
None => std::fs::read_to_string(&abs_path)
.map_err(|e| anyhow!("failed to read file '{rel_path}': {e}"))?,
};
let manager = ctx
.lsp_manager
@@ -1,5 +1,5 @@
use anyhow::Result;
use serde_json::{json, Value};
use serde_json::Value;
use std::fmt::Write;
use crate::tool::{Tool, ToolCtx};
@@ -7,73 +7,43 @@ use crate::tool::{Tool, ToolCtx};
pub struct LspReferences;
impl Tool for LspReferences {
fn name(&self) -> &'static str {
"lsp_references"
}
fn name(&self) -> &'static str { "lsp_references" }
fn description(&self) -> &'static str {
"Find all references to a symbol at a cursor position. \
`server` is optional if omitted, the server is auto-detected from the file's extension."
}
fn parameters(&self) -> Value {
json!({
"type": "object",
"properties": {
"server": {
"type": "string",
"description": "Name of the connected LSP server. Optional — auto-detected from the file extension if omitted."
},
"path": {
"type": "string",
"description": "Path to the file (relative to workspace root)"
},
"line": {
"type": "integer",
"description": "Line number (0-based)"
},
"column": {
"type": "integer",
"description": "Column number (0-based)"
}
},
"required": ["path", "line", "column"]
})
}
fn parameters(&self) -> Value { super::lsp_cursor_params(false) }
fn run(&self, ctx: &ToolCtx, args: &Value) -> Result<String> {
let result = super::run_lsp_query(ctx, args, |client, uri, line, column| {
let (ref_result, _line, _column) = super::run_lsp_query(ctx, args, None, |client, uri, line, column| {
client.references(uri, line, column)
});
})?;
match result {
Ok((ref_result, _line, _column)) => {
let locations = ref_result.as_array().cloned().unwrap_or_default();
if locations.is_empty() {
return Ok("No references found for this symbol.".to_string());
}
let mut output = format!("{} reference(s) found:\n", locations.len());
for (i, loc) in locations.iter().enumerate().take(50) {
let target_uri = loc.get("uri").and_then(|u| u.as_str()).unwrap_or("?");
let range = loc.get("range").and_then(|r| r.get("start"));
let rl = range
.and_then(|s| s.get("line"))
.and_then(serde_json::Value::as_i64)
.unwrap_or(0);
let rc = range
.and_then(|s| s.get("character"))
.and_then(serde_json::Value::as_i64)
.unwrap_or(0);
let path_str = target_uri.strip_prefix("file://").unwrap_or(target_uri);
writeln!(output, " {}. {}:{}:{}", i + 1, path_str, rl + 1, rc + 1).unwrap();
}
if locations.len() > 50 {
writeln!(output, " ... and {} more references", locations.len() - 50).unwrap();
}
Ok(output)
}
Err(e) => Err(e),
let locations = ref_result.as_array().cloned().unwrap_or_default();
if locations.is_empty() {
return Ok("No references found for this symbol.".to_string());
}
let mut output = format!("{} reference(s) found:\n", locations.len());
for (i, loc) in locations.iter().enumerate().take(50) {
let target_uri = loc.get("uri").and_then(|u| u.as_str()).unwrap_or("?");
let range = loc.get("range").and_then(|r| r.get("start"));
let rl = range
.and_then(|s| s.get("line"))
.and_then(serde_json::Value::as_i64)
.unwrap_or(0);
let rc = range
.and_then(|s| s.get("character"))
.and_then(serde_json::Value::as_i64)
.unwrap_or(0);
let path_str = target_uri.strip_prefix("file://").unwrap_or(target_uri);
writeln!(output, " {}. {}:{}:{}", i + 1, path_str, rl + 1, rc + 1).unwrap();
}
if locations.len() > 50 {
writeln!(output, " ... and {} more references", locations.len() - 50).unwrap();
}
Ok(output)
}
}
+50
View File
@@ -1,6 +1,7 @@
//! Tool trait, execution context, and the registry of all built-in tools.
use anyhow::Result;
use serde_json::Value;
use sha2::Digest;
use std::path::PathBuf;
use std::sync::atomic::AtomicBool;
use std::sync::{Arc, Mutex};
@@ -237,6 +238,55 @@ pub fn tool_defs(tools: &[Box<dyn Tool>]) -> Vec<crate::dto::provider::request::
.collect()
}
/// After a successful write/edit tool run, compute content hash and byte
/// delta, then persist an `EditLogEntry` to the session's edit log.
///
/// Used by both the main agent turn loop (`turn.rs`) and the subagent engine
/// (`engine.rs`) to avoid duplicating the SHA-256 / bytes_delta / entry
/// construction / save sequence.
pub fn log_write_edit_tool(
args: &serde_json::Value,
tool_name: &str,
origin_tag: &str,
session_dir: &std::path::Path,
session_id: &str,
) {
let reason = args
.get("reason")
.and_then(|v| v.as_str())
.unwrap_or("unnamed");
let path = args
.get("path")
.and_then(|v| v.as_str())
.unwrap_or("unknown");
let content = args.get("content").or_else(|| args.get("new"));
let content_str = content.and_then(|v| v.as_str()).unwrap_or("");
let content_sha256 = hex::encode(sha2::Sha256::digest(content_str.as_bytes()));
let bytes_delta = if tool_name == "write" {
content_str.len() as i64
} else {
let old = args.get("old").and_then(|v| v.as_str()).unwrap_or("");
let new = args.get("new").and_then(|v| v.as_str()).unwrap_or("");
(new.len() as i64 - old.len() as i64).abs()
};
let entry = zesdex_cms::domain::edit_log::EditLogEntry {
ts: chrono::Utc::now().timestamp_millis(),
tool: tool_name.to_string(),
path: path.to_string(),
reason: reason.to_string(),
content_sha256,
bytes_delta,
origin: origin_tag.to_string(),
session_id: session_id.to_string(),
};
use zesdex_cms::domain::repository::EditLogRepository;
let repo =
zesdex_cms::infrastructure::persistence::edit_log_repo::JsonlEditLogRepository::new();
if let Ok(mut el) = repo.open(session_dir) {
let _ = repo.append(session_dir, &mut el, entry);
}
}
/// Extract a required string argument from a JSON args map.
///
/// Return: the value as `String` if present and a string type; `Err` if missing
+2 -10
View File
@@ -45,18 +45,10 @@ impl Tool for Cd {
let path = super::super::resolve_path(&ctx.workspaces, &rel)?;
if !path.exists() {
return Ok(format!(
"path '{}' does not exist (resolved to {})",
rel,
path.display()
));
return Ok(super::path_not_found(&rel, &path));
}
if !path.is_dir() {
return Ok(format!(
"path '{}' is not a directory (resolved to {})",
rel,
path.display()
));
return Ok(super::path_not_a_directory(&rel, &path));
}
let canon = path.canonicalize().unwrap_or(path);
@@ -56,11 +56,7 @@ impl Tool for DirCacheUpdate {
let path = super::super::resolve_path(&ctx.workspaces, &rel)?;
if !path.exists() {
return Ok(format!(
"path '{}' does not exist (resolved to {})",
rel,
path.display()
));
return Ok(super::path_not_found(&rel, &path));
}
let entries = walk_directory(&path);
@@ -57,18 +57,10 @@ impl Tool for DirList {
let path = super::super::resolve_path(&ctx.workspaces, &rel)?;
if !path.exists() {
return Ok(format!(
"path '{}' does not exist (resolved to {})",
rel,
path.display()
));
return Ok(super::path_not_found(&rel, &path));
}
if !path.is_dir() {
return Ok(format!(
"path '{}' is not a directory (resolved to {})",
rel,
path.display()
));
return Ok(super::path_not_a_directory(&rel, &path));
}
let entries: Vec<String> = fs::read_dir(&path)
@@ -1,7 +1,28 @@
//! Small standalone utility tools (cd, dir listing/caching, pong, todowrite).
use std::path::Path;
pub mod cd;
pub mod dir_cache_update;
pub mod dir_list;
pub mod pong;
pub mod todofinish;
pub mod todowrite;
/// Format a "path does not exist" message.
pub fn path_not_found(rel: &str, path: &Path) -> String {
format!(
"path '{}' does not exist (resolved to {})",
rel,
path.display()
)
}
/// Format a "path is not a directory" message.
pub fn path_not_a_directory(rel: &str, path: &Path) -> String {
format!(
"path '{}' is not a directory (resolved to {})",
rel,
path.display()
)
}
@@ -5,8 +5,7 @@
//!
//! Why: gives callers a cheap, dependency-free way to verify the tool
//! harness is reachable and responding before running real work.
use super::super::Tool;
use super::super::ToolCtx;
use super::super::{Tool, ToolCtx};
use anyhow::Result;
use serde_json::{json, Value};
@@ -14,9 +13,7 @@ use serde_json::{json, Value};
pub struct Pong;
impl Tool for Pong {
fn name(&self) -> &'static str {
"pong"
}
fn name(&self) -> &'static str { "pong" }
fn description(&self) -> &'static str {
"Simple connectivity check. Echoes back any input for health checks and latency testing."
@@ -8,9 +8,7 @@ use std::path::PathBuf;
pub struct Todofinish;
impl Tool for Todofinish {
fn name(&self) -> &'static str {
"todofinish"
}
fn name(&self) -> &'static str { "todofinish" }
fn description(&self) -> &'static str {
"Mark tasks as finished in the session todo list (todo.md). You can mark all tasks as finished by leaving the 'task_index' empty, or specify a 1-based index to finish a specific task."
-6
View File
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Chat transcript panel rendering — tight inline log style.
//!
//! Flow: `draw_chat` turns `state.transcript_cache.messages` into a dense,
-6
View File
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Top-level TUI render pipeline: layouts the terminal into chat / input
//! / status regions, dispatches overlay rendering with glassmorphism-style
//! centered panels, and floats toast notifications over the top-right corner.
@@ -1,4 +1,4 @@
use ratatui::style::{Modifier, Style};
use ratatui::style::Style;
use ratatui::text::{Line, Span};
use ratatui::widgets::{Block, Paragraph};
use ratatui::Frame;
@@ -10,14 +10,7 @@ pub fn render(
block: Block<'static>,
state: &crate::app::state::rest::AppStateRest,
) {
let block = block
.title(Span::styled(
" Bash Jobs ",
Style::default()
.fg(Theme::ACCENT_ORANGE)
.add_modifier(Modifier::BOLD),
))
.border_style(Style::default().fg(Theme::ACCENT_ORANGE));
let block = super::overlay_block(block, "Bash Jobs", Theme::ACCENT_ORANGE);
let lines: Vec<Line> = state
.session_runtime
.as_ref()
@@ -1,5 +1,4 @@
use ratatui::style::{Modifier, Style};
use ratatui::text::Span;
use ratatui::style::Style;
use ratatui::widgets::{Block, Paragraph, Wrap};
use ratatui::Frame;
use crate::view::theme::Theme;
@@ -10,14 +9,7 @@ pub fn render(
block: Block<'static>,
_state: &crate::app::state::rest::AppStateRest,
) {
let block = block
.title(Span::styled(
" Help ",
Style::default()
.fg(Theme::INFO)
.add_modifier(Modifier::BOLD),
))
.border_style(Style::default().fg(Theme::INFO));
let block = super::overlay_block(block, "Help", Theme::INFO);
let content = crate::prompts::HELP_TEXT;
let paragraph = Paragraph::new(content)
.block(block)
+19 -1
View File
@@ -19,11 +19,29 @@ pub mod todo;
pub mod usage;
use ratatui::layout::Rect;
use ratatui::style::Style;
use ratatui::style::{Modifier, Style};
use ratatui::text::Span;
use ratatui::widgets::{Block, Borders, Clear};
use ratatui::Frame;
use super::theme::Theme;
/// Decorate an overlay block with a styled title and matching border color.
///
/// Every overlay renders a `Block` with a title bar in its variant colour
/// and a matching border — this helper centralises the `Span::styled` +
/// `border_style` boilerplate that was duplicated identically in 14 overlay
/// modules.
pub fn overlay_block(block: Block<'static>, title: &str, color: ratatui::style::Color) -> Block<'static> {
block
.title(Span::styled(
format!(" {title} "),
Style::default()
.fg(color)
.add_modifier(Modifier::BOLD),
))
.border_style(Style::default().fg(color))
}
/// Compute a centered rectangle within `area` at the given percentage width
/// and height. The result is always at least 40 cols wide and 10 rows tall.
pub fn centered_rect(area: Rect, percent_x: u16, percent_y: u16) -> Rect {
@@ -1,4 +1,4 @@
use ratatui::style::{Modifier, Style};
use ratatui::style::Style;
use ratatui::text::{Line, Span};
use ratatui::widgets::{Block, Paragraph};
use ratatui::Frame;
@@ -10,14 +10,7 @@ pub fn render(
block: Block<'static>,
state: &crate::app::state::rest::AppStateRest,
) {
let block = block
.title(Span::styled(
" Settings ",
Style::default()
.fg(Theme::PRIMARY)
.add_modifier(Modifier::BOLD),
))
.border_style(Style::default().fg(Theme::PRIMARY));
let block = super::overlay_block(block, "Settings", Theme::PRIMARY);
let lines = vec![
Line::from(Span::styled(
format!(" Provider: {}", state.settings.provider),
-6
View File
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Status bar rendering for the TUI — modern segmented bar design.
//!
//! Flow: `draw_status_bar` reads live connection/turn state off
@@ -3,13 +3,6 @@
//! `ConversationServiceImpl` is generic over `R: ConversationRepository`,
//! delegating all persistence to that adapter.
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
use anyhow::{Context, Result};
use crate::domain::conversation::{ChatMessage, Conversation};
@@ -3,13 +3,6 @@
//! `MemoryServiceImpl` is generic over `R: MemoryRepository`, delegating
//! all persistence to that adapter.
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
use anyhow::{Context, Result};
use crate::domain::memory::Memory;
-7
View File
@@ -3,13 +3,6 @@
//! Each service is generic over its repository trait so the concrete
//! persistence adapter is injected at composition root.
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
pub mod conversation_service;
pub mod memory_service;
pub mod settings_service;
@@ -3,13 +3,6 @@
//! `SettingsServiceImpl` is generic over `S: SettingsRepository` and
//! `C: AppConfigRepository`, delegating all persistence to those adapters.
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
use anyhow::Result;
use crate::domain::app_config::{AppConfig, ProviderConfig};
@@ -5,13 +5,6 @@
//! This is a pure data structure with **no I/O logic**. Load/save
//! responsibilities live in [`AppConfigRepository`](super::repository::AppConfigRepository).
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
use std::collections::HashMap;
use serde::{Deserialize, Serialize};
-7
View File
@@ -4,13 +4,6 @@
//! This is a pure data structure with **no I/O logic**. Load/save
//! responsibilities live in [`EditLogRepository`](super::repository::EditLogRepository).
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
use serde::{Deserialize, Serialize};
/// A single recorded file edit: which tool made it, to which path, why,
-7
View File
@@ -5,13 +5,6 @@
//! This is a pure data structure with **no I/O logic**. Load/save
//! responsibilities live in [`MemoryRepository`](super::repository::MemoryRepository).
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
use std::path::{Path, PathBuf};
use serde::{Deserialize, Serialize};
-7
View File
@@ -3,13 +3,6 @@
//! This layer has zero infrastructure dependencies; all I/O is expressed through
//! repository traits defined in [`repository`].
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
pub mod app_config;
pub mod conversation;
pub mod edit_log;
@@ -4,13 +4,6 @@
//! adapters implement. The domain and application layers depend only on
//! these traits, never on concrete persistence implementations.
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
use std::path::Path;
use anyhow::Result;
-7
View File
@@ -3,13 +3,6 @@
//! These traits are implemented by the application layer and consumed by
//! infrastructure adapters (HTTP handlers, CLI commands, etc.).
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
use anyhow::Result;
use super::conversation::{ChatMessage, Conversation};
-7
View File
@@ -5,13 +5,6 @@
//! This is a pure data structure with **no I/O logic**. Load/save
//! responsibilities live in [`SettingsRepository`](super::repository::SettingsRepository).
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
use std::collections::HashMap;
use serde::{Deserialize, Serialize};
@@ -4,13 +4,6 @@
//! They are independent of the domain entities so the API contract can
//! evolve without coupling to the domain model.
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
use serde::{Deserialize, Serialize};
// ---------------------------------------------------------------------------
@@ -5,13 +5,6 @@
//! HTTP framework — callers (e.g. Axum routes) are responsible for mapping
//! `Result` into HTTP responses.
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
use anyhow::{Context, Result};
use crate::domain::memory::Memory;
@@ -1,12 +1,5 @@
//! HTTP adapter — handler functions and DTOs for CMS REST endpoints.
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
pub mod dto;
pub mod handlers;
@@ -2,12 +2,5 @@
//!
//! Contains persistence implementations (file I/O) and HTTP handler adapters.
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
pub mod http;
pub mod persistence;
@@ -5,18 +5,11 @@
//! On load, auto-detects Claude credentials from the environment or
//! `~/.claude/settings.json` and merges them into the provider map.
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
use std::io::Write;
use std::path::Path;
use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use zesdex_utils::write_json_atomic;
use crate::domain::app_config::{AppConfig, ModelRole, ProviderConfig};
use crate::domain::repository::AppConfigRepository;
@@ -134,31 +127,8 @@ impl AppConfigRepository for JsonAppConfigRepository {
std::fs::create_dir_all(base_dir)
.with_context(|| format!("failed to create base dir '{}'", base_dir.display()))?;
let path = base_dir.join("app_config.json");
let tmp = base_dir.join("app_config.json.tmp");
let json =
serde_json::to_string_pretty(config).context("failed to serialize app config")?;
{
let mut f = std::fs::OpenOptions::new()
.create(true)
.truncate(true)
.write(true)
.open(&tmp)
.with_context(|| format!("failed to write temp file '{}'", tmp.display()))?;
f.write_all(json.as_bytes())?;
f.sync_all()?;
}
std::fs::rename(&tmp, &path).with_context(|| {
format!(
"failed to rename '{}' -> '{}'",
tmp.display(),
path.display()
)
})?;
if let Some(parent) = path.parent() {
if let Ok(d) = std::fs::File::open(parent) {
let _ = d.sync_all();
}
}
write_json_atomic(&path, config, None)
.with_context(|| "failed to save app_config")?;
tracing::debug!("app_config saved to '{}'", path.display());
Ok(())
}
@@ -4,17 +4,10 @@
//!
//! Uses write-then-rename with fsync for crash safety.
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
use std::io::Write;
use std::path::Path;
use anyhow::{Context, Result};
use zesdex_utils::write_json_atomic;
use crate::domain::conversation::Conversation;
use crate::domain::repository::ConversationRepository;
@@ -44,31 +37,8 @@ impl ConversationRepository for JsonConversationRepository {
std::fs::create_dir_all(session_dir)
.with_context(|| format!("failed to create session dir '{}'", session_dir.display()))?;
let path = session_dir.join("conversation.json");
let tmp = session_dir.join("conversation.json.tmp");
let json = serde_json::to_string_pretty(conversation)
.context("failed to serialize conversation")?;
{
let mut f = std::fs::OpenOptions::new()
.create(true)
.truncate(true)
.write(true)
.open(&tmp)
.with_context(|| format!("failed to write temp file '{}'", tmp.display()))?;
f.write_all(json.as_bytes())?;
f.sync_all()?;
}
std::fs::rename(&tmp, &path).with_context(|| {
format!(
"failed to rename '{}' -> '{}'",
tmp.display(),
path.display()
)
})?;
if let Some(parent) = path.parent() {
if let Ok(d) = std::fs::File::open(parent) {
let _ = d.sync_all();
}
}
write_json_atomic(&path, conversation, None)
.with_context(|| "failed to save conversation")?;
tracing::debug!("conversation saved to '{}'", path.display());
Ok(())
}
@@ -5,13 +5,6 @@
//! Append-only log: new entries are appended to the file, never rewritten.
//! In-memory cache is capped at 10K entries to prevent unbounded growth.
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
use std::io::{BufRead, BufReader, Write};
use std::path::Path;
@@ -7,13 +7,6 @@
//! name, description, kind, created_at, updated_at, lifecycle,
//! outcome, scope, before, after, provenances
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
use std::collections::HashMap;
use std::io::Write;
use std::path::Path;
@@ -1,12 +1,5 @@
//! Persistence adapters — concrete file-based repository implementations.
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
pub mod app_config_repo;
pub mod conversation_repo;
pub mod edit_log_repo;
@@ -4,17 +4,10 @@
//!
//! Uses write-then-rename with fsync for crash safety.
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
use std::io::Write;
use std::path::Path;
use anyhow::{Context, Result};
use zesdex_utils::write_json_atomic;
use crate::domain::repository::SettingsRepository;
use crate::domain::settings::Settings;
@@ -56,31 +49,8 @@ impl SettingsRepository for JsonSettingsRepository {
std::fs::create_dir_all(base_dir)
.with_context(|| format!("failed to create base dir '{}'", base_dir.display()))?;
let path = base_dir.join("settings.json");
let tmp = base_dir.join("settings.json.tmp");
let json =
serde_json::to_string_pretty(settings).context("failed to serialize settings")?;
{
let mut f = std::fs::OpenOptions::new()
.create(true)
.truncate(true)
.write(true)
.open(&tmp)
.with_context(|| format!("failed to write temp file '{}'", tmp.display()))?;
f.write_all(json.as_bytes())?;
f.sync_all()?;
}
std::fs::rename(&tmp, &path).with_context(|| {
format!(
"failed to rename '{}' -> '{}'",
tmp.display(),
path.display()
)
})?;
if let Some(parent) = path.parent() {
if let Ok(d) = std::fs::File::open(parent) {
let _ = d.sync_all();
}
}
write_json_atomic(&path, settings, None)
.with_context(|| "failed to save settings")?;
tracing::debug!("settings saved to '{}'", path.display());
Ok(())
}
+1
View File
@@ -18,3 +18,4 @@ url.workspace = true
reqwest.workspace = true
tokio.workspace = true
tracing.workspace = true
zesdex-utils.workspace = true
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Session metadata: id, title, workspace roots, and message/token counts,
//! persisted as `session.json` per session directory.
use chrono::Utc;
@@ -58,26 +52,19 @@ impl Session {
/// Persist this session's metadata to `session.json`, atomically
/// with fsync for crash safety.
///
/// Flow: ensure the session directory exists → serialize to pretty
/// JSON → write to `session.json.tmp` → fsync → rename over
/// `session.json` → fsync parent directory.
/// Flow: ensure the session directory exists → atomically write
/// pretty-printed JSON via `write_json_atomic`.
///
/// Why: write-then-rename avoids a torn/partial `session.json` if
/// interrupted mid-write; fsync before rename ensures the data is
/// on disk before the rename makes it visible.
///
/// Return: `Ok(())` on success, or an `io::Error` from any step.
pub fn save(&self, base_dir: &Path) -> std::io::Result<()> {
/// Return: `Ok(())` on success, or an `anyhow::Error` from any step.
pub fn save(&self, base_dir: &Path) -> anyhow::Result<()> {
let dir = self.session_dir(base_dir);
std::fs::create_dir_all(&dir)?;
let path = dir.join("session.json");
let data = serde_json::to_string_pretty(self)?;
let tmp = dir.join("session.json.tmp");
std::fs::write(&tmp, data)?;
let f = std::fs::File::open(&tmp)?;
f.sync_all()?;
std::fs::rename(&tmp, path)?;
let _ = std::fs::File::open(&dir).and_then(|d| d.sync_all());
zesdex_utils::write_json_atomic(&path, self, None)?;
Ok(())
}
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! PID-file based advisory lock preventing two processes from operating on
//! the same session directory concurrently.
use std::fs;
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! In-memory conversation state: message history plus the system prompt and
//! model parameters used to drive the LLM.
use serde::{Deserialize, Serialize};
@@ -77,20 +71,14 @@ impl Conversation {
/// Persist the conversation to a JSON file at the given base directory.
///
/// Flow: compute path from `session_id` → ensure directory exists →
/// serialize to pretty JSON → write-then-rename with fsync.
/// atomically write pretty-printed JSON via `write_json_atomic`.
///
/// Return: `Ok(())` on success, or an `io::Error` from any step.
pub fn save_conversation(&self, base_dir: &std::path::Path) -> std::io::Result<()> {
/// Return: `Ok(())` on success, or an `anyhow::Error` from any step.
pub fn save_conversation(&self, base_dir: &std::path::Path) -> anyhow::Result<()> {
let dir = base_dir.join("sessions").join(&self.session_id);
std::fs::create_dir_all(&dir)?;
let path = dir.join("conversation.json");
let data = serde_json::to_string_pretty(self)?;
let tmp = dir.join("conversation.json.tmp");
std::fs::write(&tmp, data)?;
let f = std::fs::File::open(&tmp)?;
f.sync_all()?;
std::fs::rename(&tmp, path)?;
let _ = std::fs::File::open(&dir).and_then(|d| d.sync_all());
zesdex_utils::write_json_atomic(&path, self, None)?;
Ok(())
}
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Chat message types shared across the entity layer: `Role` and `ChatMessage`
//! with convenience constructors.
use serde::{Deserialize, Serialize};
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Provider-facing DTOs: chat completion request, response, streaming types,
//! and the SSE stream parser.
use serde::{Deserialize, Serialize};
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Filesystem layout for zesdex's persistent and scratch data directories.
use serde::{Deserialize, Serialize};
use std::path::PathBuf;
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Tool-call DTOs embedded in assistant chat messages.
//!
//! Flow: provider response/stream carries `tool_calls` on an assistant
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Record of one completed tool invocation, kept for transcript/history.
use serde::{Deserialize, Serialize};
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Token usage accounting shared by streaming and non-streaming responses.
use serde::{Deserialize, Serialize};
+7
View File
@@ -1,3 +1,10 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Domain entity types for the Zesdex application.
//!
//! This crate contains ALL domain entity types as pure data structures
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! OAuth flow use-cases.
//!
//! `OAuthServiceImpl` drives the authorization-code + PKCE flow:
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Session management use-cases.
//!
//! `SessionServiceImpl` implements `SessionService` by delegating to
-6
View File
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Pure OAuth entities — no HTTP or persistence logic.
use serde::{Deserialize, Serialize};
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Repository trait definitions (pure — no impls, no concrete persistence).
use std::path::Path;
-6
View File
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Service trait definitions — use-case interfaces for session management
//! and OAuth flows.
use crate::domain::oauth::{OAuthConfig, OAuthToken};
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! IAM-specific HTTP / IPC DTOs (Data Transfer Objects).
use serde::{Deserialize, Serialize};
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! IPC / HTTP handler functions.
//!
//! Each handler is a plain function that takes a service reference and a
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Minimal loopback HTTP server for capturing OAuth authorization-code redirects.
//!
//! Ported from `zesdex-backend::service::oauth::loopback` to centralise OAuth
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Filesystem-backed `OAuthRepository` implementation.
//!
//! Tokens are stored as a single JSON file. Writes use a write-then-rename
@@ -11,6 +5,8 @@
//! 0o600 on Unix.
use std::path::Path;
use zesdex_utils::write_json_atomic;
use crate::domain::oauth::OAuthToken;
use crate::domain::repository::OAuthRepository;
@@ -30,20 +26,7 @@ impl OAuthRepository for FileSystemOAuthRepository {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
let data = serde_json::to_string_pretty(token)?;
let tmp = path.with_extension("tmp");
std::fs::write(&tmp, data)?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(&tmp, std::fs::Permissions::from_mode(0o600))?;
}
let f = std::fs::File::open(&tmp)?;
f.sync_all()?;
std::fs::rename(&tmp, path)?;
if let Some(parent) = path.parent() {
let _ = std::fs::File::open(parent).and_then(|d| d.sync_all());
}
write_json_atomic(path, token, Some(0o600))?;
Ok(())
}
@@ -1,15 +1,11 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Filesystem-backed `SessionRepository` implementation.
//!
//! Each session is stored as `<base_dir>/sessions/<id>/session.json`.
//! Writes use a write-then-rename + fsync pattern for crash safety.
use std::path::Path;
use zesdex_utils::write_json_atomic;
use crate::domain::repository::SessionRepository;
use crate::domain::session::Session;
@@ -61,17 +57,7 @@ impl SessionRepository for FileSystemSessionRepository {
let dir = session.session_dir(base_dir);
std::fs::create_dir_all(&dir)?;
let path = dir.join("session.json");
let data = serde_json::to_string_pretty(session)?;
let tmp = dir.join("session.json.tmp");
std::fs::write(&tmp, data)?;
// fsync before rename ensures the data is on disk.
let f = std::fs::File::open(&tmp)?;
f.sync_all()?;
std::fs::rename(&tmp, path)?;
// fsync the parent directory so the rename survives a crash.
if let Some(parent) = dir.parent() {
let _ = std::fs::File::open(parent).and_then(|d| d.sync_all());
}
write_json_atomic(&path, session, None)?;
Ok(())
}
-6
View File
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! SQLite database connection pool initialisation and schema migrations.
//!
//! Uses `r2d2` + `r2d2_sqlite` for connection pooling with the same
-6
View File
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! JWT token utilities for HMAC-SHA256 / HS256 signing and verification.
//!
//! Uses the `jsonwebtoken` crate under the hood.
-6
View File
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Argon2 password hashing and verification utilities.
//!
//! Uses the `argon2` crate (Argon2id variant) with default parameters,
-6
View File
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Application state initialisation and wiring.
//!
//! This module acts as the composition root for the zesdex daemon (and
-7
View File
@@ -9,13 +9,6 @@
//! The length prefix **excludes** itself — it encodes only the number of
//! payload bytes that follow.
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
use anyhow::{Context, Result};
use std::io::{Read, Write};
-7
View File
@@ -5,13 +5,6 @@
//! [`DaemonFrame`] are serialised as JSON messages framed with a
//! length prefix (see [`crate::frame`]).
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
use serde::{Deserialize, Serialize};
// ---------------------------------------------------------------------------
-7
View File
@@ -4,13 +4,6 @@
//! [`IpcServer`] wraps a [`UnixListener`] and provides a blocking
//! `accept` method that returns a [`Connection`] for each new client.
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
use crate::conn::Connection;
use anyhow::{Context, Result};
use std::os::unix::net::UnixListener;
+62 -62
View File
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! Authentication middleware — session-lock based auth for Axum.
//!
//! Provides:
@@ -121,6 +115,52 @@ pub struct SessionAuthMiddleware<S> {
store: Arc<Store>,
}
// ---------------------------------------------------------------------------
// Session ID helpers
// ---------------------------------------------------------------------------
/// Extract and validate `X-Session-Id` from request headers.
///
/// Flow: read header -> validate non-empty -> return ID or a 401 error response.
fn extract_session_id<ReqBody>(req: &Request<ReqBody>) -> Result<String, Response> {
let session_id = req
.headers()
.get("X-Session-Id")
.and_then(|v| v.to_str().ok())
.map(|s| s.to_string());
match session_id {
Some(id) if !id.is_empty() => Ok(id),
_ => Err((StatusCode::UNAUTHORIZED, "missing X-Session-Id header").into_response()),
}
}
/// Validate session and build identity from request context.
///
/// Flow: validate session in store -> extract User-Agent -> build SessionIdentity.
fn validate_and_build_identity<ReqBody>(
session_id: &str,
store: &Store,
req: &Request<ReqBody>,
) -> Result<SessionIdentity, Response> {
match validate_session(session_id, store) {
Ok(_session) => {
let user_agent = req
.headers()
.get(header::USER_AGENT)
.and_then(|v| v.to_str().ok())
.unwrap_or("")
.to_string();
Ok(SessionIdentity::new(session_id.to_string(), user_agent))
}
Err(e) => Err((
StatusCode::UNAUTHORIZED,
format!("session validation failed: {e}"),
)
.into_response()),
}
}
impl<S, ReqBody> Service<Request<ReqBody>> for SessionAuthMiddleware<S>
where
S: Service<Request<ReqBody>, Response = Response> + Send + 'static,
@@ -139,39 +179,17 @@ where
fn call(&mut self, mut req: Request<ReqBody>) -> Self::Future {
let store = Arc::clone(&self.store);
// Extract session id from header.
let session_id = req
.headers()
.get("X-Session-Id")
.and_then(|v| v.to_str().ok())
.map(|s| s.to_string());
let session_id = match session_id {
Some(id) if !id.is_empty() => id,
_ => {
let resp =
(StatusCode::UNAUTHORIZED, "missing X-Session-Id header").into_response();
return Box::pin(async move { Ok(resp) });
}
let session_id = match extract_session_id(&req) {
Ok(id) => id,
Err(resp) => return Box::pin(async move { Ok(resp) }),
};
// Validate session.
if let Err(e) = validate_session(&session_id, &store) {
let resp = (
StatusCode::UNAUTHORIZED,
format!("session validation failed: {e}"),
)
.into_response();
return Box::pin(async move { Ok(resp) });
}
let user_agent = req
.headers()
.get(header::USER_AGENT)
.and_then(|v| v.to_str().ok())
.unwrap_or("")
.to_string();
let identity = SessionIdentity::new(session_id, user_agent);
req.extensions_mut().insert(identity);
match validate_and_build_identity(&session_id, &store, &req) {
Ok(identity) => {
req.extensions_mut().insert(identity);
}
Err(resp) => return Box::pin(async move { Ok(resp) }),
};
let fut = self.inner.call(req);
Box::pin(fut)
@@ -192,33 +210,15 @@ pub async fn require_session(
mut req: Request<axum::body::Body>,
next: axum::middleware::Next,
) -> Response {
let session_id = req
.headers()
.get("X-Session-Id")
.and_then(|v| v.to_str().ok())
.map(|s| s.to_string());
let session_id = match session_id {
Some(id) if !id.is_empty() => id,
_ => {
return (StatusCode::UNAUTHORIZED, "missing X-Session-Id header").into_response();
}
let session_id = match extract_session_id(&req) {
Ok(id) => id,
Err(resp) => return resp,
};
if let Err(e) = validate_session(&session_id, &store) {
return (
StatusCode::UNAUTHORIZED,
format!("session validation failed: {e}"),
)
.into_response();
}
let user_agent = req
.headers()
.get(header::USER_AGENT)
.and_then(|v| v.to_str().ok())
.unwrap_or("")
.to_string();
let identity = SessionIdentity::new(session_id, user_agent);
let identity = match validate_and_build_identity(&session_id, &store, &req) {
Ok(identity) => identity,
Err(resp) => return resp,
};
req.extensions_mut().insert(identity);
next.run(req).await
}
-6
View File
@@ -1,9 +1,3 @@
#![allow(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_precision_loss,
clippy::cast_possible_wrap
)]
//! CORS layer factory for the daemon HTTP (IPC) server.
//!
//! Since the daemon only listens on `127.0.0.1`, the CORS policy is
+46
View File
@@ -0,0 +1,46 @@
//! Crash-safe atomic file write helper.
//!
//! Writes serializable data to a temp file, fsyncs, then renames into
//! place to guarantee atomicity. On Unix, an optional `mode` sets the
//! permissions of the final file (e.g. `0o600` for OAuth tokens).
use std::io::Write;
use std::path::Path;
use serde::Serialize;
/// Atomically write serializable `data` to `path`.
///
/// Flow: serialize to pretty JSON -> write to `path.tmp` -> fsync -> rename -> fsync parent.
/// If `mode` is `Some`, set permissions before rename (Unix only).
///
/// Edge case: tmp file name uses `with_extension("tmp")` which replaces
/// the existing extension -- correct for `foo.json` -> `foo.tmp`. For paths
/// without an extension (unlikely in this codebase), appends `.tmp`.
pub fn write_json_atomic<T: Serialize>(path: &Path, data: &T, mode: Option<u32>) -> anyhow::Result<()> {
let tmp = path.with_extension("tmp");
let bytes = serde_json::to_vec_pretty(data)?;
{
let mut f = std::fs::OpenOptions::new()
.create(true)
.truncate(true)
.write(true)
.open(&tmp)?;
f.write_all(&bytes)?;
f.sync_all()?;
}
if let Some(m) = mode {
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(&tmp, std::fs::Permissions::from_mode(m))?;
}
#[cfg(not(unix))]
{ let _ = m; }
}
std::fs::rename(&tmp, path)?;
if let Some(parent) = path.parent() {
let _ = std::fs::File::open(parent).and_then(|d| d.sync_all());
}
Ok(())
}

Some files were not shown because too many files have changed in this diff Show More