MythEclipse/zesdex
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(token): add refresh token verification to TokenService

Browse Source 
feat(bootstrap): create temporary settings and config files to prevent data loss

refactor(edit_log): switch from Vec to VecDeque for efficient memory management

fix(gateway): ensure store directories are created before starting the API server

refactor(bgbash): implement a global singleton for BashControl

feat(auth): enhance session authentication middleware to use SessionRepository

fix(edit_log_repo): update to use VecDeque for in-memory edit log storage

fix(memory_repo): add newline escaping for frontmatter fields

fix(session_lock_repo): improve error handling for lock file operations

fix(bash_tools): prevent path traversal in job_id argument

refactor(delete): enforce empty directory deletion in file system tools

fix(edit): optimize string replacement to only replace the first occurrence

fix(git_cred): improve credential management with piped input to git commands

feat(git_operator): add safety filter to block destructive git operations

fix(shell): register background jobs in Bash control

feat(spawn): add access tier specification for pipeline stages

refactor(hive_mind): run directives concurrently for improved performance

fix(auth): update refresh token verification in the refresh handler

fix(chat): optimize LLM client usage based on model matching

fix(conversations): enhance message deletion to target specific indices

feat(api): add JWT authentication middleware for all API routes

fix(state): implement refresh token verification in JwtTokenService

fix(daemon): improve usage tracking with saturating addition

fix(tui): handle compacted messages in the TUI state management
This commit is contained in:
asepharyana
2026-07-20 12:26:10 +07:00
parent 600ea041ef
commit a04651905f
26 changed files with 497 additions and 453 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/infrastructure/src/bgbash/control.rs
+9 -1
View File
@@ -1,12 +1,20 @@
//! Background bash control — list, cancel, and inspect background processes.
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::sync::{Arc, Mutex, OnceLock};
use tracing::error;
use super::job::BashJob;
/// Global accessor for the shared BashControl singleton.
///
/// Used by the Bash tool (to register jobs) and BashKill (to look them up).
pub fn bash_control() -> &'static BashControl {
static BASH_CONTROL: OnceLock<BashControl> = OnceLock::new();
BASH_CONTROL.get_or_init(BashControl::new)
}
/// Central registry of all running background bash jobs.
pub struct BashControl {
jobs: Mutex<HashMap<String, Arc<BashJob>>>,
apps/infrastructure/src/middleware/auth.rs
+48 -26
View File
@@ -1,7 +1,12 @@
//! Authentication middleware — session-lock based auth for Axum.
//!
//! Validates `X-Session-Id` header against the `SessionRepository` before
//! forwarding the request to the inner service.
use std::future::Future;
use std::path::PathBuf;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use axum::body::Body;
@@ -9,6 +14,7 @@ use axum::http::{Request, Response, StatusCode};
use axum::response::IntoResponse;
use serde::{Deserialize, Serialize};
use tower::{Layer, Service};
use zesdex_domain::auth::{SessionId, SessionRepository};
/// Identity extracted from a validated session.
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -30,40 +36,50 @@ impl SessionIdentity {
}
/// Tower Layer that produces SessionAuthMiddleware services.
///
/// Holds a reference to the `SessionRepository` and the base directory
/// needed to validate session IDs.
#[derive(Debug, Clone)]
pub struct SessionAuthLayer;
pub struct SessionAuthLayer<R: SessionRepository + Send + Sync + 'static> {
base_dir: PathBuf,
repo: Arc<R>,
}
impl SessionAuthLayer {
pub fn new() -> Self {
Self
impl<R: SessionRepository + Send + Sync + 'static> SessionAuthLayer<R> {
pub fn new(base_dir: PathBuf, repo: Arc<R>) -> Self {
Self { base_dir, repo }
}
}
impl Default for SessionAuthLayer {
fn default() -> Self {
Self
}
}
impl<S> Layer<S> for SessionAuthLayer {
type Service = SessionAuthMiddleware<S>;
impl<S, R> Layer<S> for SessionAuthLayer<R>
where
R: SessionRepository + Send + Sync + 'static,
{
type Service = SessionAuthMiddleware<S, R>;
fn layer(&self, inner: S) -> Self::Service {
SessionAuthMiddleware { inner }
SessionAuthMiddleware {
inner,
base_dir: self.base_dir.clone(),
repo: self.repo.clone(),
}
}
}
/// Tower Service that validates X-Session-Id before forwarding.
#[derive(Debug, Clone)]
pub struct SessionAuthMiddleware<S> {
pub struct SessionAuthMiddleware<S, R: SessionRepository + Send + Sync + 'static> {
inner: S,
base_dir: PathBuf,
repo: Arc<R>,
}
impl<S, ReqBody> Service<Request<ReqBody>> for SessionAuthMiddleware<S>
impl<S, ReqBody, R> Service<Request<ReqBody>> for SessionAuthMiddleware<S, R>
where
S: Service<Request<ReqBody>, Response = Response<Body>> + Send + 'static,
S::Future: Send + 'static,
ReqBody: Send + 'static,
R: SessionRepository + Send + Sync + 'static,
{
type Response = S::Response;
type Error = S::Error;
@@ -81,18 +97,24 @@ where
.and_then(|v| v.to_str().ok())
.map(|s| s.to_string());
if session_id.as_deref() != Some("valid-session") {
// In production, this validates against the store
return Box::pin(async move {
Ok((
StatusCode::UNAUTHORIZED,
"missing or invalid X-Session-Id header",
)
.into_response())
});
// Validate the session against the repository.
match session_id {
Some(sid) => match SessionId::new(&sid) {
Ok(id) => match self.repo.load_session(&self.base_dir, &id) {
Ok(_session) => {
// Session is valid — forward the request.
let fut = self.inner.call(req);
return Box::pin(fut);
}
Err(_) => { /* fall through to 401 */ }
},
Err(_) => { /* fall through to 401 */ }
},
None => { /* fall through to 401 */ }
}
let fut = self.inner.call(req);
Box::pin(fut)
Box::pin(async move {
Ok((StatusCode::UNAUTHORIZED, "missing or invalid X-Session-Id header").into_response())
})
}
}
apps/infrastructure/src/persistence/cms/edit_log_repo.rs
+9 -8
View File
@@ -1,6 +1,7 @@
//! JSONL file–backed `EditLogRepository`.
//! Stores `EditLog` as an append-only newline-delimited JSON file.
use std::collections::VecDeque;
use std::io::{BufRead, BufReader, Write};
use std::path::Path;
@@ -18,21 +19,21 @@ impl JsonlEditLogRepository {
Self
}
fn load_from_disk(path: &Path) -> Vec<EditLogEntry> {
fn load_from_disk(path: &Path) -> VecDeque<EditLogEntry> {
let Ok(file) = std::fs::File::open(path) else {
return Vec::new();
return VecDeque::new();
};
let reader = BufReader::new(file);
let mut entries: Vec<EditLogEntry> = Vec::new();
let mut entries: VecDeque<EditLogEntry> = VecDeque::new();
for line in reader.lines() {
let Ok(line) = line else {
continue;
};
if let Ok(entry) = serde_json::from_str::<EditLogEntry>(&line) {
if entries.len() >= MAX_MEMORY_ENTRIES {
entries.remove(0);
entries.pop_front();
}
entries.push(entry);
entries.push_back(entry);
}
}
entries
@@ -74,14 +75,14 @@ impl EditLogRepository for JsonlEditLogRepository {
file.write_all(line.as_bytes())?;
file.sync_all()?;
}
log.entries.push(entry);
log.entries.push_back(entry);
if log.entries.len() > MAX_MEMORY_ENTRIES {
log.entries.remove(0);
log.entries.pop_front();
}
Ok(())
}
fn entries(&self, log: &EditLog) -> Vec<EditLogEntry> {
log.entries.clone()
log.entries.clone().into_iter().collect()
}
}
apps/infrastructure/src/persistence/cms/memory_repo.rs
+35 -8
View File
@@ -17,26 +17,37 @@ impl MarkdownMemoryRepository {
Self
}
/// Escape newlines in field values so they do not break the
/// line-oriented frontmatter parser.
fn escape_newlines(s: &str) -> String {
s.replace('\n', "\\n")
}
/// Unescape `\n` back to actual newlines after frontmatter parsing.
fn unescape_newlines(s: &str) -> String {
s.replace("\\n", "\n")
}
fn build_frontmatter(memory: &Memory) -> String {
let outcome_line = memory
.outcome
.as_ref()
.map(|o| format!("outcome: {o}\n"))
.map(|o| format!("outcome: {}\n", Self::escape_newlines(o)))
.unwrap_or_default();
let scope_line = memory
.scope
.as_ref()
.map(|s| format!("scope: {s}\n"))
.map(|s| format!("scope: {}\n", Self::escape_newlines(s)))
.unwrap_or_default();
let before_line = memory
.before_snippet
.as_ref()
.map(|s| format!("before: {s}\n"))
.map(|s| format!("before: {}\n", Self::escape_newlines(s)))
.unwrap_or_default();
let after_line = memory
.after_snippet
.as_ref()
.map(|s| format!("after: {s}\n"))
.map(|s| format!("after: {}\n", Self::escape_newlines(s)))
.unwrap_or_default();
let prov_line = if memory.provenances.is_empty() {
String::new()
@@ -101,14 +112,30 @@ impl MarkdownMemoryRepository {
.get("updated_at")
.and_then(|v| v.parse().ok())
.unwrap_or(0),
outcome: front.get("outcome").cloned().filter(|s| !s.is_empty()),
outcome: front
.get("outcome")
.cloned()
.filter(|s| !s.is_empty())
.map(|s| Self::unescape_newlines(&s)),
lifecycle: front
.get("lifecycle")
.cloned()
.unwrap_or_else(|| "new".to_string()),
scope: front.get("scope").cloned().filter(|s| !s.is_empty()),
before_snippet: front.get("before").cloned().filter(|s| !s.is_empty()),
after_snippet: front.get("after").cloned().filter(|s| !s.is_empty()),
scope: front
.get("scope")
.cloned()
.filter(|s| !s.is_empty())
.map(|s| Self::unescape_newlines(&s)),
before_snippet: front
.get("before")
.cloned()
.filter(|s| !s.is_empty())
.map(|s| Self::unescape_newlines(&s)),
after_snippet: front
.get("after")
.cloned()
.filter(|s| !s.is_empty())
.map(|s| Self::unescape_newlines(&s)),
provenances: front
.get("provenances")
.cloned()
apps/infrastructure/src/persistence/iam/session_lock_repo.rs
+9 -5
View File
@@ -36,7 +36,7 @@ impl SessionLockRepository for FileSystemSessionLockRepository {
Err(e) => return Err(RepositoryError::Io(e)),
}
let content = std::fs::read_to_string(&path).unwrap_or_default();
let content = std::fs::read_to_string(&path).map_err(RepositoryError::Io)?;
if let Ok(existing_pid) = content.trim().parse::<u32>() {
if self.is_alive(existing_pid) {
return Ok(false);
@@ -46,10 +46,14 @@ impl SessionLockRepository for FileSystemSessionLockRepository {
let tmp = path.with_extension("lock.tmp");
{
let mut tmp_file = std::fs::OpenOptions::new()
.create(true)
.truncate(true)
.create_new(true)
.write(true)
.open(&tmp)?;
.open(&tmp)
.map_err(|_| {
RepositoryError::Other(
"another process is replacing the lock".to_string(),
)
})?;
write!(tmp_file, "{pid}")?;
tmp_file.sync_all()?;
}
@@ -62,7 +66,7 @@ impl SessionLockRepository for FileSystemSessionLockRepository {
fn unlock(&self, session_dir: &Path) -> Result<(), RepositoryError> {
let path = session_dir.join(".lock");
let _ = std::fs::remove_file(path);
std::fs::remove_file(path)?;
Ok(())
}
apps/infrastructure/src/tools/bash_tools.rs
+9 -16
View File
@@ -37,6 +37,11 @@ impl Tool for BashOutput {
let job_id = arg_str(args, "job_id")?;
info!("Getting output for job: {job_id}");
// Prevent path traversal
if job_id.contains('/') || job_id.contains('\\') || job_id.contains("..") {
anyhow::bail!("invalid job_id '{job_id}': must not contain path separators");
}
// Read from the session's bash output directory
let output_dir = ctx.session_dir.join("bash-outputs");
let output_file = output_dir.join(&job_id);
@@ -80,23 +85,11 @@ impl Tool for BashKill {
fn run(&self, _ctx: &ToolCtx, args: &Value) -> Result<String> {
let job_id = crate::tools::arg_str(args, "job_id")?;
info!("bash_kill called for job: {job_id}");
// Try to kill by PID (if job_id is numeric) or by process name
if let Ok(pid) = job_id.parse::<u32>() {
use std::process::Command;
match Command::new("kill").arg(pid.to_string()).output() {
Ok(output) if output.status.success() => {
Ok(format!("Killed background job '{job_id}' (PID {pid})"))
}
Ok(output) => {
let stderr = String::from_utf8_lossy(&output.stderr);
Ok(format!("Failed to kill job '{job_id}': {stderr}"))
}
Err(e) => {
Ok(format!("Failed to kill job '{job_id}': {e}"))
}
}
if crate::bgbash::control::bash_control().cancel(&job_id) {
Ok(format!("Killed background job '{job_id}'"))
} else {
Ok(format!("Invalid job ID '{job_id}' — expected numeric PID"))
anyhow::bail!("no active background job found with ID '{job_id}'")
}
}
}
apps/infrastructure/src/tools/fs/delete.rs
+4 -2
View File
@@ -45,8 +45,10 @@ impl Tool for Delete {
fs::remove_file(&path)?;
Ok(format!("Deleted file '{rel}'"))
} else if path.is_dir() {
fs::remove_dir_all(&path)?;
Ok(format!("Deleted directory '{rel}' and all contents"))
fs::remove_dir(&path).map_err(|e| {
anyhow::anyhow!("failed to delete directory '{rel}': {e} (directory must be empty)")
})?;
Ok(format!("Deleted empty directory '{rel}'"))
} else {
anyhow::bail!("'{rel}' is neither a file nor a directory")
}
apps/infrastructure/src/tools/fs/edit.rs
+1 -1
View File
@@ -56,7 +56,7 @@ impl Tool for Edit {
anyhow::bail!("old text not found in '{}'", rel);
}
let new_content = content.replace(&old, &new);
let new_content = content.replacen(&old, &new, 1);
fs::write(&path, &new_content)?;
Ok(format!(
apps/infrastructure/src/tools/git/git_cred.rs
+20 -5
View File
@@ -3,6 +3,8 @@
use crate::tools::{execute_cmd, Tool, ToolCtx};
use anyhow::Result;
use serde_json::{json, Value};
use std::io::Write;
use std::process::{Command, Stdio};
pub struct GitCred;
@@ -49,10 +51,15 @@ impl Tool for GitCred {
let url = crate::tools::arg_str(args, "url")?;
let username = crate::tools::arg_str(args, "username")?;
let password = crate::tools::arg_str(args, "password")?;
let _input = format!("url={url}\nusername={username}\npassword={password}\n");
let _output = execute_cmd(
std::process::Command::new("git").args(["credential", "approve"]),
)?;
let input = format!("url={url}\nusername={username}\npassword={password}\n");
let mut child = Command::new("git")
.args(["credential", "approve"])
.stdin(Stdio::piped())
.spawn()?;
if let Some(ref mut stdin) = child.stdin {
stdin.write_all(input.as_bytes())?;
}
child.wait()?;
Ok(format!("Credential stored for {url}"))
}
"list" => {
@@ -63,7 +70,15 @@ impl Tool for GitCred {
}
"erase" => {
let url = crate::tools::arg_str(args, "url")?;
let _input = format!("url={url}\n");
let input = format!("url={url}\n");
let mut child = Command::new("git")
.args(["credential", "reject"])
.stdin(Stdio::piped())
.spawn()?;
if let Some(ref mut stdin) = child.stdin {
stdin.write_all(input.as_bytes())?;
}
child.wait()?;
Ok(format!("Credential erased for {url}"))
}
_ => anyhow::bail!("unknown action: {}", action),
apps/infrastructure/src/tools/git/git_operator.rs
+7
View File
@@ -1,5 +1,6 @@
//! Git operator tool — commit, push, pull, branch operations.
use crate::tools::shell_filter::git::check_git_destructive;
use crate::tools::{execute_cmd, Tool, ToolCtx};
use anyhow::Result;
use serde_json::{json, Value};
@@ -46,6 +47,12 @@ impl Tool for GitOperator {
})
.unwrap_or_default();
// Safety filter: block destructive git operations
let cmd_str = format!("git {} {}", operation, extra_args.join(" "));
if let Err(e) = check_git_destructive(&cmd_str) {
anyhow::bail!("blocked: {e}");
}
let mut cmd = std::process::Command::new("git");
cmd.arg(&operation);
for arg in &extra_args {
apps/infrastructure/src/tools/shell.rs
+1
View File
@@ -61,6 +61,7 @@ impl Tool for Bash {
if run_in_background {
let job = crate::bgbash::job::spawn_bash_job(cmd);
crate::bgbash::control::bash_control().register(job.clone());
return Ok(format!("Background job: {}", job.id));
}
apps/infrastructure/src/tools/spawn.rs
+15 -3
View File
@@ -151,7 +151,8 @@ impl Tool for SpawnPipeline {
"items": {
"type": "object",
"properties": {
"directive": {"type": "string", "description": "Directive for this pipeline stage"}
"directive": {"type": "string", "description": "Directive for this pipeline stage"},
"access": {"type": "string", "enum": ["read", "write", "full"], "description": "Access tier for this stage"}
},
"required": ["directive"]
},
@@ -200,17 +201,28 @@ impl Tool for SpawnPipeline {
.unwrap_or("")
.to_string();
let access_str = stage
.get("access")
.and_then(|v| v.as_str())
.unwrap_or("full");
let access = match access_str {
"read" => AccessTier::Read,
"write" => AccessTier::Write,
_ => AccessTier::Full,
};
let subagent_ctx = SubagentContext::new(
directive.clone(),
ctx.clone(),
"full".to_string(),
access_str.to_string(),
base_url.clone(),
api_key.clone(),
model.clone(),
);
let result = rt.block_on(async {
run_agent(subagent_ctx, &directive, AccessTier::Full, ctx.clone()).await
run_agent(subagent_ctx, &directive, access, ctx.clone()).await
})?;
pipeline_result.push_str(&format!("Stage {}: {}\n", i, result));
apps/infrastructure/src/workflow/hive_mind/cycle.rs
+36 -22
View File
@@ -1,9 +1,10 @@
//! Hive-mind cycle execution — run one cycle of parallel nodes.
//!
//! Flow: load settings → resolve LLM credentials → for each directive,
//! build a SubagentContext and call run_agent → collect NodeOutputs.
//! Flow: load settings → resolve LLM credentials → run all directives in the
//! cycle concurrently via try_join_all → collect Vec<NodeOutput>.
use anyhow::Result;
use futures_util::future::try_join_all;
use tracing::info;
use zesdex_domain::cms::{AppConfigRepository, SettingsRepository};
@@ -21,8 +22,9 @@ use crate::workflow::hive_mind::types::{CognitiveCycle, NodeOutput};
/// Flow:
/// 1. Load `Settings` and `AppConfig` from the store directory.
/// 2. Resolve provider, model, base_url, and api_key.
/// 3. For each directive → build `SubagentContext` → `run_agent` (Full access).
/// 4. Collect `NodeOutput` results.
/// 3. Spawn all directives concurrently — each builds a `SubagentContext`
/// and calls `run_agent` (Full access).
/// 4. `try_join_all` waits for all to complete, then collect `NodeOutput`s.
pub async fn execute_cycle(
cycle: &CognitiveCycle,
tool_ctx: &ToolCtx,
@@ -52,25 +54,37 @@ pub async fn execute_cycle(
let api_key = crate::llm::provider::resolve_api_key(&settings, &app_config);
let mut outputs = Vec::new();
for (i, directive) in cycle.directives.iter().enumerate() {
let ctx = SubagentContext::new(
directive.clone(),
tool_ctx.clone(),
"full".to_string(),
base_url.clone(),
api_key.clone(),
model.clone(),
);
let cycle_index = cycle.index;
let result = run_agent(ctx, directive, AccessTier::Full, tool_ctx.clone()).await?;
// Run all directives in this cycle concurrently.
let handles: Vec<_> = cycle
.directives
.iter()
.enumerate()
.map(|(i, directive)| {
let dir = directive.clone();
let ctx = SubagentContext::new(
dir.clone(),
tool_ctx.clone(),
"full".to_string(),
base_url.clone(),
api_key.clone(),
model.clone(),
);
let tc = tool_ctx.clone();
outputs.push(NodeOutput {
id: format!("Node-{}-{}", cycle.index, i),
directive: directive.clone(),
output: result,
});
}
async move {
let result = run_agent(ctx, &dir, AccessTier::Full, tc).await?;
Ok::<NodeOutput, anyhow::Error>(NodeOutput {
id: format!("Node-{}-{}", cycle_index, i),
directive: dir,
output: result,
})
}
})
.collect();
Ok(outputs)
let results = try_join_all(handles).await?;
Ok(results)
}