asepharyana
e080d9fc6b
Perilaku should_shape/shape_messages tidak berubah, hanya sumber penghitungan token yang sekarang lewat context::tokens (tiktoken-rs) menggantikan heuristik char/3 bawaannya sendiri.
1841 lines
81 KiB
Rust
1841 lines
81 KiB
Rust
//! The `Action` enum and its single dispatcher, `apply_action` — the
|
|
//! chokepoint through which every key input, streaming event, and async
|
|
//! background-thread result mutates `AppStateRest`.
|
|
//!
|
|
//! Flow: controllers/subagent threads construct `Action` values → the event
|
|
//! loop calls `apply_action(&mut state, action)` → for turn-producing
|
|
//! actions (`SubmitInput`), `spawn_turn` is kicked off on a background OS
|
|
//! thread which drives `run_agent_turn` (stream to the LLM, gate and
|
|
//! execute tool calls via `Harness`, archive messages to `SQLite`, log edits)
|
|
//! and pushes `TurnEvent`s onto a shared queue → on the next `Tick`, queued
|
|
//! `TurnEvent`s are drained back into `AppStateRest` (transcript, toasts,
|
|
//! usage counters).
|
|
//!
|
|
//! Why: keeping all state mutation behind one function means callers only
|
|
//! need to know how to *produce* actions, not how to update state safely;
|
|
//! 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)]
|
|
|
|
use std::collections::VecDeque;
|
|
use std::fmt::Write;
|
|
|
|
use crate::app::harness::Verdict;
|
|
use sha2::Digest;
|
|
use crate::app::review::{should_trigger_review, trigger_review};
|
|
use crate::app::state::rest::{AppStateRest, ChatMessageDisplay};
|
|
use crate::app::state::runtime::TurnEvent;
|
|
use crate::app::state::types::{Origin, Overlay, Toast, ToastKind};
|
|
use crate::dto::chat::message::{ChatMessage, Role};
|
|
|
|
/// A single, well-typed event in the app — produced by key input, the
|
|
/// streaming pipeline, or subagent threads — that mutates `AppStateRest`
|
|
/// when applied via `apply_action`.
|
|
///
|
|
/// Step bounds intentionally left unbounded (`usize::MAX`) so the agent can
|
|
/// continue across as many turns as needed. Each iteration still honours
|
|
/// `tc.abort_flag` and the per-call LLM timeout, so a runaway loop is
|
|
/// observable and cancellable from the UI.
|
|
#[derive(Debug, Clone)]
|
|
pub enum Action {
|
|
ForceQuit,
|
|
SubmitInput(String),
|
|
DeleteChar,
|
|
DeleteCharRight,
|
|
CursorLeft,
|
|
CursorRight,
|
|
HistoryUp,
|
|
HistoryDown,
|
|
ScrollUp,
|
|
ScrollDown,
|
|
OpenOverlay(Overlay),
|
|
CloseOverlay,
|
|
SystemNote {
|
|
kind: String,
|
|
message: String,
|
|
},
|
|
QuitConfirm,
|
|
Resize(u16, u16),
|
|
Tick,
|
|
|
|
LessonAccept {
|
|
name: String,
|
|
},
|
|
LessonReject {
|
|
name: String,
|
|
},
|
|
LessonDelete {
|
|
name: String,
|
|
},
|
|
StartOAuth {
|
|
provider: String,
|
|
},
|
|
OpenEditor {
|
|
path: String,
|
|
},
|
|
McpAdd {
|
|
name: String,
|
|
command: String,
|
|
},
|
|
ModelList,
|
|
AbortTurn,
|
|
Compact,
|
|
|
|
}
|
|
|
|
/// Apply an `Action` to the application state.
|
|
///
|
|
/// Flow: pattern-match the variant → mutate `state` (input buffer, scroll
|
|
/// position, overlay, transcript, runtime, toasts, dirty flag, etc.) →
|
|
/// for `Tick`, also drain queued `TurnEvent`s and run periodic side jobs
|
|
/// (staleness sweep, pending-lesson commit).
|
|
///
|
|
/// Why: the single chokepoint that turns every typed key and async event
|
|
/// into a state change, so callers (controllers, subagent threads) only
|
|
/// need to know how to *produce* actions.
|
|
///
|
|
/// Return: nothing; `state` is mutated in place.
|
|
#[allow(clippy::too_many_lines)]
|
|
pub fn apply_action(state: &mut AppStateRest, action: Action) {
|
|
match action {
|
|
Action::ForceQuit => {
|
|
save_current_session(state);
|
|
state.shutdown_lsp();
|
|
state.quit = true;
|
|
}
|
|
|
|
Action::SubmitInput(text) => {
|
|
state.input.submit();
|
|
let text = text.trim().to_string();
|
|
if text.is_empty() {
|
|
state.dirty = true;
|
|
return;
|
|
}
|
|
state.push_transcript(ChatMessageDisplay::new(Role::User, text.clone()));
|
|
if let Some(ref mut rt) = state.session_runtime {
|
|
rt.push_message(ChatMessage::user(text));
|
|
refresh_lesson_counters(&state.memory_dir, rt);
|
|
} else {
|
|
let _ = std::fs::create_dir_all(&state.memory_dir);
|
|
}
|
|
state.misc.thinking = true;
|
|
spawn_turn(state);
|
|
state.dirty = true;
|
|
}
|
|
Action::DeleteChar => {
|
|
state.input.delete_left();
|
|
state.dirty = true;
|
|
}
|
|
Action::DeleteCharRight => {
|
|
state.input.delete_right();
|
|
state.dirty = true;
|
|
}
|
|
Action::CursorLeft => {
|
|
state.input.char_left();
|
|
}
|
|
Action::CursorRight => {
|
|
state.input.char_right();
|
|
}
|
|
Action::HistoryUp => {
|
|
state.input.history_up();
|
|
state.dirty = true;
|
|
}
|
|
Action::HistoryDown => {
|
|
state.input.history_down();
|
|
state.dirty = true;
|
|
}
|
|
Action::ScrollUp => {
|
|
state.scroll.scroll_up(5);
|
|
state.dirty = true;
|
|
}
|
|
Action::ScrollDown => {
|
|
state.scroll.scroll_down(5);
|
|
state.dirty = true;
|
|
}
|
|
Action::OpenOverlay(overlay) => {
|
|
state.misc.overlay = overlay;
|
|
if overlay == Overlay::Learning || overlay == Overlay::Rewind || overlay == Overlay::ModelSelector {
|
|
state.misc.selected_index = 0;
|
|
}
|
|
state.dirty = true;
|
|
}
|
|
Action::OpenEditor { path } => {
|
|
let resolved = crate::tool::resolve_path(&state.workspace_roots, &path);
|
|
match resolved {
|
|
Ok(abs_path) => {
|
|
let content = std::fs::read_to_string(&abs_path)
|
|
.unwrap_or_default();
|
|
let lines: Vec<String> = content.lines().map(std::string::ToString::to_string).collect();
|
|
let ed = crate::app::mode::editor::EditorState::open(
|
|
abs_path.to_string_lossy().to_string(),
|
|
Some(lines),
|
|
);
|
|
state.misc.editor = Some(ed);
|
|
state.misc.overlay = Overlay::Editor;
|
|
state.push_toast(Toast::new(ToastKind::Info, format!("Editing {path}")));
|
|
}
|
|
Err(e) => {
|
|
state.push_toast(Toast::new(ToastKind::Error, format!("Failed to open {path}: {e}")));
|
|
}
|
|
}
|
|
state.dirty = true;
|
|
}
|
|
Action::McpAdd { name, command } => {
|
|
let extra_args: Vec<String> = command.split_whitespace().map(std::string::ToString::to_string).collect();
|
|
let cmd = extra_args.first().cloned().unwrap_or_default();
|
|
let args: Vec<String> = extra_args.into_iter().skip(1).collect();
|
|
match state.mcp_manager.connect_stdio(&name, &cmd, &args) {
|
|
Ok(()) => {
|
|
let tool_count = state.mcp_manager.servers.last()
|
|
.map_or(0, |s| s.tools.len());
|
|
state.push_toast(Toast::new(ToastKind::Success,
|
|
format!("Connected MCP server '{name}' ({tool_count} tools)")));
|
|
state.dirty = true;
|
|
}
|
|
Err(e) => {
|
|
state.push_toast(Toast::new(ToastKind::Error,
|
|
format!("MCP connect failed: {e}")));
|
|
}
|
|
}
|
|
}
|
|
Action::ModelList => {
|
|
state.misc.selected_index = 0;
|
|
state.misc.overlay = Overlay::ModelSelector;
|
|
state.dirty = true;
|
|
}
|
|
Action::CloseOverlay => {
|
|
// If the overlay is the Editor, dismiss it properly first
|
|
if state.misc.overlay == Overlay::Editor {
|
|
crate::app::mode::editor::handle_editor_dismiss(state);
|
|
}
|
|
state.misc.overlay = Overlay::None;
|
|
state.dirty = true;
|
|
}
|
|
Action::SystemNote { kind: _kind, message } => {
|
|
let toast = crate::app::state::types::Toast::new(
|
|
crate::app::state::types::ToastKind::Info,
|
|
message,
|
|
);
|
|
state.push_toast(toast);
|
|
}
|
|
Action::QuitConfirm => {
|
|
state.misc.overlay = Overlay::QuitConfirm;
|
|
state.dirty = true;
|
|
}
|
|
Action::Resize(w, _h) => {
|
|
state.scroll.set_max_visible(w as usize);
|
|
state.dirty = true;
|
|
}
|
|
|
|
Action::StartOAuth { provider } => {
|
|
let turn_events = state.turn_events.clone();
|
|
let provider_clone = provider.clone();
|
|
std::thread::spawn(move || {
|
|
let result = run_oauth_flow(&provider_clone);
|
|
let message = match result {
|
|
Ok(msg) => msg,
|
|
Err(e) => format!("OAuth login failed: {e}"),
|
|
};
|
|
if let Ok(mut q) = turn_events.lock() {
|
|
q.push_back(TurnEvent::SystemNote {
|
|
kind: "oauth".to_string(),
|
|
message,
|
|
});
|
|
}
|
|
});
|
|
let toast = Toast::new(ToastKind::Info, format!("Opening browser for {provider} login..."));
|
|
state.push_toast(toast);
|
|
state.dirty = true;
|
|
}
|
|
Action::Tick => {
|
|
state.misc.tick_count = state.misc.tick_count.wrapping_add(1);
|
|
let now_ms = chrono::Utc::now().timestamp_millis();
|
|
state.misc.drain_expired_toasts(now_ms);
|
|
|
|
if state.misc.tick_count.is_multiple_of(10) {
|
|
let todo_path = state.session_dir.join("todo.md");
|
|
if let Ok(content) = std::fs::read_to_string(&todo_path) {
|
|
if content != state.misc.todo_content {
|
|
state.misc.todo_content = content;
|
|
state.dirty = true;
|
|
}
|
|
} else if !state.misc.todo_content.is_empty() {
|
|
state.misc.todo_content.clear();
|
|
state.dirty = true;
|
|
}
|
|
}
|
|
|
|
// Background API connectivity check — runs on a background thread
|
|
// every ~1s while disconnected, every ~30s while connected, so the
|
|
// status bar reflects real API availability without user input.
|
|
let check_interval = if state.misc.api_connected { 600 } else { 20 };
|
|
if state.misc.tick_count.is_multiple_of(check_interval) {
|
|
spawn_api_connectivity_check(state);
|
|
}
|
|
crate::app::review::maybe_run_staleness_sweep(state);
|
|
if let Some(ref rt) = state.session_runtime {
|
|
let _ = crate::app::review::process_pending_lessons(&rt.session_dir, &state.memory_dir);
|
|
}
|
|
|
|
// Drain LSP provision progress messages into toast notifications.
|
|
// Collect messages under the lock, then push toasts outside it to avoid
|
|
// a borrow-conflict with state.push_toast (which also accesses state).
|
|
let pending: Vec<String> = state.lsp_provision_msgs.lock()
|
|
.ok()
|
|
.map(|mut q| q.drain(..).collect())
|
|
.unwrap_or_default();
|
|
for msg in &pending {
|
|
let kind = if msg.contains("not available") || msg.contains("failed") {
|
|
ToastKind::Warning
|
|
} else if msg.contains("connected") || msg.contains("✓") {
|
|
ToastKind::Success
|
|
} else {
|
|
ToastKind::Info
|
|
};
|
|
state.push_toast(Toast::new(kind, msg.clone()));
|
|
}
|
|
|
|
let events: Vec<TurnEvent> = {
|
|
if let Ok(mut q) = state.turn_events.lock() {
|
|
q.drain(..).collect()
|
|
} else {
|
|
Vec::new()
|
|
}
|
|
};
|
|
let mut turn_finished = false;
|
|
for event in events {
|
|
match event {
|
|
TurnEvent::AssistantMessage(msg) => {
|
|
state.misc.thinking = false;
|
|
state.misc.api_connected = true;
|
|
let display_content = msg.content.clone().unwrap_or_default();
|
|
if !display_content.is_empty() {
|
|
state.push_transcript(ChatMessageDisplay::new(Role::Assistant, display_content));
|
|
}
|
|
if let Some(ref mut rt) = state.session_runtime {
|
|
rt.push_message(msg);
|
|
}
|
|
}
|
|
TurnEvent::ToolResult { tool_call_id, tool_name, output, is_error, path } => {
|
|
state.misc.thinking = false;
|
|
let display_path = path.unwrap_or_default();
|
|
let display = if tool_name == "read" {
|
|
let line_count = output.lines().count();
|
|
if display_path.is_empty() {
|
|
format!("read: {line_count} line(s)")
|
|
} else {
|
|
format!("read: {display_path} ({line_count} lines)")
|
|
}
|
|
} else {
|
|
format!("{tool_name}: {output}")
|
|
};
|
|
state.push_transcript(ChatMessageDisplay::new(
|
|
Role::Tool,
|
|
display,
|
|
));
|
|
if let Some(ref mut rt) = state.session_runtime {
|
|
rt.push_message(ChatMessage::tool_result(tool_call_id.clone(), output.clone()));
|
|
rt.tool_call_results.push(crate::app::state::runtime::ToolCallResult {
|
|
tool_call_id,
|
|
tool_name,
|
|
output,
|
|
is_error,
|
|
duration_ms: 0,
|
|
});
|
|
}
|
|
}
|
|
TurnEvent::SystemNote { kind, message } => {
|
|
if kind == "edits" {
|
|
if let Some(ref mut rt) = state.session_runtime {
|
|
if let Ok(count) = message.parse::<u32>() {
|
|
rt.edit_count += count;
|
|
}
|
|
}
|
|
if should_trigger_review(state, Origin::Main) {
|
|
trigger_review(state);
|
|
}
|
|
} else if kind == "review" {
|
|
state.misc.lesson_running = false;
|
|
let counted = if let Some(ref mut rt) = state.session_runtime {
|
|
refresh_lesson_counters(&state.memory_dir, rt);
|
|
true
|
|
} else {
|
|
false
|
|
};
|
|
if let Some(ref mut rt) = state.session_runtime {
|
|
if counted {
|
|
rt.consecutive_empty_reviews = 0;
|
|
} else {
|
|
rt.consecutive_empty_reviews += 1;
|
|
}
|
|
}
|
|
state.push_toast(Toast::new(ToastKind::Info, message));
|
|
} else if kind == "task_retry" {
|
|
state.push_transcript(ChatMessageDisplay::new(
|
|
crate::dto::chat::message::Role::System,
|
|
message.clone(),
|
|
));
|
|
state.push_toast(Toast::new(ToastKind::Info, "Auto-continuing unfinished tasks...".to_string()));
|
|
if let Some(ref mut rt) = state.session_runtime {
|
|
rt.push_message(crate::dto::chat::message::ChatMessage::system(message.clone()));
|
|
}
|
|
} else if kind == "connectivity" {
|
|
state.misc.api_connected = message == "connected";
|
|
} else if kind == "hive_mind_converged" {
|
|
if let Some(ref mut rt) = state.session_runtime {
|
|
rt.hive_mind_converged = true;
|
|
}
|
|
} else if kind == "pipeline" {
|
|
// Clear old workflow agents when a new pipeline starts.
|
|
if message == HIVE_MIND_KICKOFF_NOTE {
|
|
state.workflow_engine.agents.clear();
|
|
state.workflow_engine.findings.clear();
|
|
}
|
|
// popup removed, no overlay to reset
|
|
state.push_toast(Toast {
|
|
kind: ToastKind::Info,
|
|
message: message.clone(),
|
|
created_at: chrono::Utc::now().timestamp_millis(),
|
|
lifetime_ms: 12000,
|
|
});
|
|
state.dirty = true;
|
|
} else if kind == "bg-test-gen" {
|
|
let escalated = message.starts_with("ESCALATED:");
|
|
state.push_toast(Toast {
|
|
kind: if escalated { ToastKind::Error } else { ToastKind::Info },
|
|
message: message.clone(),
|
|
created_at: chrono::Utc::now().timestamp_millis(),
|
|
lifetime_ms: if escalated { 30000 } else { 8000 },
|
|
});
|
|
state.dirty = true;
|
|
} else if kind == "bg-arch-review" || kind == "bg-security-review" {
|
|
let escalated = message.starts_with("ESCALATED:");
|
|
state.push_toast(Toast {
|
|
kind: if escalated { ToastKind::Error } else { ToastKind::Info },
|
|
message: message.clone(),
|
|
created_at: chrono::Utc::now().timestamp_millis(),
|
|
lifetime_ms: if escalated { 30000 } else { 10000 },
|
|
});
|
|
state.dirty = true;
|
|
} else if kind == "workflow_done" {
|
|
state.push_toast(Toast {
|
|
kind: ToastKind::Success,
|
|
message: message.clone(),
|
|
created_at: chrono::Utc::now().timestamp_millis(),
|
|
lifetime_ms: 10000,
|
|
});
|
|
state.push_transcript(ChatMessageDisplay::new(
|
|
crate::dto::chat::message::Role::System,
|
|
format!("✓ {message}"),
|
|
));
|
|
// overlay removed
|
|
state.dirty = true;
|
|
} else if kind == "workflow_error" {
|
|
state.push_toast(Toast {
|
|
kind: ToastKind::Error,
|
|
message: message.clone(),
|
|
created_at: chrono::Utc::now().timestamp_millis(),
|
|
lifetime_ms: 12000,
|
|
});
|
|
state.push_transcript(ChatMessageDisplay::new(
|
|
crate::dto::chat::message::Role::System,
|
|
format!("✗ {message}"),
|
|
));
|
|
// overlay removed
|
|
state.dirty = true;
|
|
} else {
|
|
state.push_toast(Toast::new(ToastKind::Info, message));
|
|
}
|
|
}
|
|
TurnEvent::StreamStart => {
|
|
state.misc.thinking = false;
|
|
state.misc.api_connected = true;
|
|
state.push_transcript(ChatMessageDisplay::new(Role::Assistant, String::new()));
|
|
}
|
|
TurnEvent::StreamToken(delta) => {
|
|
if let Some(last) = state.transcript_cache.messages.last_mut() {
|
|
if last.role == Role::Assistant {
|
|
last.content.push_str(&delta);
|
|
state.transcript_cache.dirty = true;
|
|
}
|
|
}
|
|
}
|
|
TurnEvent::StreamDone(msg) => {
|
|
state.misc.thinking = false;
|
|
if let Some(ref mut rt) = state.session_runtime {
|
|
rt.push_message(msg);
|
|
}
|
|
}
|
|
TurnEvent::Usage { tokens_in, tokens_out } => {
|
|
if let Some(ref mut rt) = state.session_runtime {
|
|
rt.usage.tokens_in += tokens_in;
|
|
rt.usage.tokens_out += tokens_out;
|
|
rt.usage.last_tokens_in = tokens_in;
|
|
rt.usage.last_tokens_out = tokens_out;
|
|
rt.usage.api_calls += 1;
|
|
}
|
|
}
|
|
TurnEvent::ReviewUsage { tokens_in, tokens_out } => {
|
|
if let Some(ref mut rt) = state.session_runtime {
|
|
rt.usage.tokens_in += tokens_in;
|
|
rt.usage.tokens_out += tokens_out;
|
|
rt.usage.review_tokens += tokens_in + tokens_out;
|
|
rt.usage.api_calls += 1;
|
|
}
|
|
}
|
|
TurnEvent::Error(msg) => {
|
|
state.misc.api_connected = false;
|
|
let long_toast = Toast {
|
|
kind: ToastKind::Error,
|
|
message: msg.clone(),
|
|
created_at: chrono::Utc::now().timestamp_millis(),
|
|
lifetime_ms: 15000,
|
|
};
|
|
state.push_toast(long_toast);
|
|
state.push_transcript(ChatMessageDisplay::new(
|
|
crate::dto::chat::message::Role::System,
|
|
format!("Error: {msg}"),
|
|
));
|
|
turn_finished = true;
|
|
}
|
|
TurnEvent::Done => {
|
|
state.misc.thinking = false;
|
|
turn_finished = true;
|
|
}
|
|
TurnEvent::Compacted(new_msgs) => {
|
|
if let Some(ref mut rt) = state.session_runtime {
|
|
rt.messages = new_msgs;
|
|
state.push_toast(Toast::new(ToastKind::Info, "History auto-compacted by AI.".to_string()));
|
|
state.dirty = true;
|
|
}
|
|
}
|
|
TurnEvent::WorkflowAgentUpdate { agent_id, agent_name, status } => {
|
|
// Upsert the agent in the workflow engine roster.
|
|
// Running agents are pushed as new entries; status
|
|
// updates find the existing entry by id and replace it.
|
|
use crate::app::workflow::engine::WorkflowAgent;
|
|
if let Some(existing) = state.workflow_engine.agents
|
|
.iter_mut()
|
|
.find(|a| a.id == agent_id)
|
|
{
|
|
existing.status = status;
|
|
} else {
|
|
state.workflow_engine.agents.push(WorkflowAgent {
|
|
id: agent_id,
|
|
name: agent_name,
|
|
status,
|
|
});
|
|
}
|
|
// popup removed
|
|
state.dirty = true;
|
|
}
|
|
}
|
|
}
|
|
if turn_finished {
|
|
maybe_trigger_review(state);
|
|
|
|
}
|
|
if turn_finished || state.dirty {
|
|
state.dirty = true;
|
|
}
|
|
}
|
|
Action::AbortTurn => {
|
|
state.abort_flag.store(true, std::sync::atomic::Ordering::SeqCst);
|
|
state.push_toast(Toast::new(ToastKind::Warning, "Aborting generation...".to_string()));
|
|
}
|
|
Action::Compact => {
|
|
let max_wire_tokens = state.app_config.model_roles.values()
|
|
.find(|role| role.provider == state.settings.provider && role.model == state.settings.model)
|
|
.and_then(|role| role.context_window)
|
|
.unwrap_or(state.app_config.default_context_window) as usize;
|
|
|
|
if let Some(ref mut rt) = state.session_runtime {
|
|
let total_chars: usize = rt.messages.iter()
|
|
.filter_map(|m| m.content.as_deref())
|
|
.map(str::len)
|
|
.sum();
|
|
let token_estimate = total_chars / 3;
|
|
rt.messages = crate::app::runtime::context::shaping::shape_messages(&rt.messages, token_estimate, max_wire_tokens, true, None);
|
|
state.push_toast(Toast::new(ToastKind::Success, "Conversation history compacted.".to_string()));
|
|
state.dirty = true;
|
|
}
|
|
}
|
|
Action::LessonAccept { name } => {
|
|
if let Some(ref rt) = state.session_runtime {
|
|
let _ = crate::app::review::resolve_pending_lesson(
|
|
&rt.session_dir, &state.memory_dir, &name, true,
|
|
);
|
|
}
|
|
if let Some(ref mut rt) = state.session_runtime {
|
|
refresh_lesson_counters(&state.memory_dir, rt);
|
|
}
|
|
state.push_toast(Toast::new(ToastKind::Success,
|
|
format!("accepted lesson: {name}")));
|
|
state.dirty = true;
|
|
}
|
|
Action::LessonReject { name } => {
|
|
if let Some(ref rt) = state.session_runtime {
|
|
let _ = crate::app::review::resolve_pending_lesson(
|
|
&rt.session_dir, &state.memory_dir, &name, false,
|
|
);
|
|
}
|
|
if let Some(ref mut rt) = state.session_runtime {
|
|
refresh_lesson_counters(&state.memory_dir, rt);
|
|
}
|
|
state.push_toast(Toast::new(ToastKind::Info,
|
|
format!("rejected lesson: {name}")));
|
|
state.dirty = true;
|
|
}
|
|
Action::LessonDelete { name } => {
|
|
let _ = crate::model::memory::Memory::remove(&state.memory_dir, &name);
|
|
if let Some(ref mut rt) = state.session_runtime {
|
|
refresh_lesson_counters(&state.memory_dir, rt);
|
|
}
|
|
state.push_toast(Toast::new(ToastKind::Info, format!("deleted lesson: {name}")));
|
|
state.dirty = true;
|
|
}
|
|
|
|
|
|
}
|
|
}
|
|
|
|
/// Spawn a background thread that runs one full LLM turn.
|
|
///
|
|
/// Flow: check that no turn is currently in-flight → bail if so →
|
|
/// collect messages and config from state → determine API key (from
|
|
/// settings, env var, or default) → resolve generation params from
|
|
/// the current effort level → collect all tools (built-in + MCP) →
|
|
/// build `TurnCtx` → spawn a thread running `run_agent_turn` →
|
|
/// on any error, push a `TurnEvent::Error` → clear the in-flight flag
|
|
/// when the thread exits.
|
|
///
|
|
/// Why: runs on a plain OS thread so the async event loop stays responsive.
|
|
///
|
|
/// Return: nothing; results flow through `state.turn_events`.
|
|
fn spawn_turn(state: &AppStateRest) {
|
|
let in_flight = if let Ok(guard) = state.turn_in_flight.lock() {
|
|
*guard
|
|
} else {
|
|
return;
|
|
};
|
|
if in_flight {
|
|
return;
|
|
}
|
|
let messages = state
|
|
.session_runtime
|
|
.as_ref()
|
|
.map(|rt| rt.messages.clone())
|
|
.unwrap_or_default();
|
|
if messages.is_empty() {
|
|
return;
|
|
}
|
|
let mut api_key = state.settings.api_keys.get(&state.settings.provider).cloned().unwrap_or_default();
|
|
let model = state.settings.model.clone();
|
|
let base_url = state.app_config.providers.get(&state.settings.provider)
|
|
.map(|p| p.api_base.clone());
|
|
let context_window = state.app_config.model_roles.values()
|
|
.find(|role| role.provider == state.settings.provider && role.model == state.settings.model)
|
|
.and_then(|role| role.context_window)
|
|
.unwrap_or(state.app_config.default_context_window) as usize;
|
|
// The selected provider has no entry in app_config at all (e.g. the
|
|
// Claude-settings auto-detection that registers "claude" found nothing
|
|
// this run). Without this check, LlmClient::new silently falls back to
|
|
// the zen default base URL while keeping this provider's model name —
|
|
// a mismatched request that reaches a real server and comes back as a
|
|
// confusing "Missing API key" 401 from an unrelated provider, instead
|
|
// of the actual problem: the configured provider doesn't exist.
|
|
if base_url.is_none() {
|
|
if let Ok(mut q) = state.turn_events.lock() {
|
|
q.push_back(TurnEvent::Error(format!(
|
|
"Provider '{}' is not configured — no matching entry found. \
|
|
Pick a different provider in Settings, or configure it.",
|
|
state.settings.provider
|
|
)));
|
|
}
|
|
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();
|
|
}
|
|
let (temperature, max_tokens) = crate::app::mode::effort::generation_params(
|
|
state.misc.effort_level,
|
|
state.settings.max_tokens,
|
|
);
|
|
let mut tools = crate::tool::all_tools();
|
|
tools.extend(state.mcp_manager.as_tools());
|
|
let tool_defs = crate::tool::tool_defs(&tools);
|
|
let ctx = state.tool_ctx();
|
|
|
|
let edit_session_dir = state.session_dir.clone();
|
|
let session_id = state.session_id.clone();
|
|
let turn_events = state.turn_events.clone();
|
|
let in_flight_flag = state.turn_in_flight.clone();
|
|
let workspace_roots: Vec<std::path::PathBuf> = ctx.workspaces.clone();
|
|
let abort_flag = state.abort_flag.clone();
|
|
abort_flag.store(false, std::sync::atomic::Ordering::SeqCst);
|
|
let hive_mind_converged = state.session_runtime.as_ref().is_some_and(|rt| rt.hive_mind_converged);
|
|
|
|
*in_flight_flag.lock().unwrap_or_else(|e| {
|
|
tracing::error!("[spawn_turn] in_flight_flag mutex poisoned: {}", e);
|
|
e.into_inner()
|
|
}) = true;
|
|
|
|
let events_q = turn_events.clone();
|
|
|
|
std::thread::spawn(move || {
|
|
let db = crate::model::msglog::open_or_create(&edit_session_dir)
|
|
.ok()
|
|
.map(|c| std::sync::Arc::new(std::sync::Mutex::new(c)));
|
|
let tc = TurnCtx {
|
|
client: crate::service::provider::LlmClient::new(api_key, model.clone(), base_url),
|
|
tdefs: tool_defs,
|
|
tools,
|
|
ctx,
|
|
context_window,
|
|
|
|
workspace_roots,
|
|
edit_log_session_dir: edit_session_dir,
|
|
session_id,
|
|
db,
|
|
temperature,
|
|
max_tokens,
|
|
abort_flag,
|
|
hive_mind_converged,
|
|
};
|
|
let result = run_agent_turn(&tc, &messages, &events_q);
|
|
if let Err(e) = result {
|
|
if let Ok(mut q) = events_q.lock() {
|
|
q.push_back(TurnEvent::Error(e.to_string()));
|
|
}
|
|
}
|
|
if let Ok(mut flag) = in_flight_flag.lock() {
|
|
*flag = false;
|
|
}
|
|
});
|
|
}
|
|
|
|
/// Context bundle passed to `run_agent_turn` on its background thread.
|
|
struct TurnCtx {
|
|
client: crate::service::provider::LlmClient,
|
|
tdefs: Vec<crate::dto::provider::request::ToolDef>,
|
|
tools: Vec<Box<dyn crate::tool::Tool>>,
|
|
ctx: crate::tool::ToolCtx,
|
|
context_window: usize,
|
|
|
|
workspace_roots: Vec<std::path::PathBuf>,
|
|
edit_log_session_dir: std::path::PathBuf,
|
|
session_id: String,
|
|
db: Option<std::sync::Arc<std::sync::Mutex<rusqlite::Connection>>>,
|
|
temperature: f32,
|
|
max_tokens: Option<u32>,
|
|
abort_flag: std::sync::Arc<std::sync::atomic::AtomicBool>,
|
|
/// Snapshot of `SessionRuntime.hive_mind_converged` taken at the start
|
|
/// of this turn — whether a hive-mind convergence already completed
|
|
/// earlier in this session.
|
|
hive_mind_converged: bool,
|
|
}
|
|
|
|
/// 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.
|
|
///
|
|
/// Flow: list memory slugs → for each, read + parse the file → collect
|
|
/// entries whose lifecycle is not "stale" → cap total output at 3000 chars
|
|
/// to avoid dominating the prompt budget.
|
|
///
|
|
/// Why: previously, lessons existed on disk but the AI never saw them
|
|
/// unless it explicitly called `recall()`. This makes the memory system
|
|
/// actually useful by surfacing relevant knowledge automatically.
|
|
///
|
|
/// Return: a formatted string (may be empty if no memory entries exist).
|
|
fn build_memory_section(memory_dir: &std::path::Path) -> String {
|
|
let names = crate::model::memory::Memory::list(memory_dir);
|
|
if names.is_empty() {
|
|
return String::new();
|
|
}
|
|
|
|
let mut section = String::from("\n\n--- Persistent Memory ---\n");
|
|
write!(section, "Total entries: {}\n\n", names.len()).unwrap();
|
|
|
|
for name in &names {
|
|
if section.len() > 3000 {
|
|
section.push_str("... (more entries omitted, use recall() to see all)\n");
|
|
break;
|
|
}
|
|
if let Ok(mem) = crate::model::memory::Memory::read(memory_dir, name) {
|
|
if mem.lifecycle == "stale" {
|
|
continue;
|
|
}
|
|
write!(section, "## [{}] {}\n{}\n\n", mem.kind, mem.name, mem.content).unwrap();
|
|
}
|
|
}
|
|
section.push_str("---");
|
|
section
|
|
}
|
|
|
|
/// Scan `memory_dir` and update every lesson counter in `SessionRuntime`
|
|
/// from real on-disk data.
|
|
///
|
|
/// Flow: list all memory slugs → read+parse each → increment the matching
|
|
/// kind counter (user/feedback/project/reference), lifecycle counter
|
|
/// (active/stale/contradicted), and the total. If a memory cannot be read
|
|
/// (e.g. a race with deletion) it is silently skipped.
|
|
///
|
|
/// Why: previously the UI showed all zeros because nothing ever set the
|
|
/// breakdown counters. This runs on every user submit so the dashboard
|
|
/// reflects actual memory state.
|
|
fn refresh_lesson_counters(memory_dir: &std::path::Path, rt: &mut crate::app::state::runtime::SessionRuntime) {
|
|
let names = crate::model::memory::Memory::list(memory_dir);
|
|
rt.lesson_count = 0;
|
|
rt.lessons_user = 0;
|
|
rt.lessons_feedback = 0;
|
|
rt.lessons_project = 0;
|
|
rt.lessons_reference = 0;
|
|
rt.lessons_active = 0;
|
|
rt.lessons_stale = 0;
|
|
rt.lessons_contradicted = 0;
|
|
for name in &names {
|
|
if let Ok(mem) = crate::model::memory::Memory::read(memory_dir, name) {
|
|
rt.lesson_count += 1;
|
|
match mem.kind.as_str() {
|
|
"user" => rt.lessons_user += 1,
|
|
"feedback" => rt.lessons_feedback += 1,
|
|
"project" => rt.lessons_project += 1,
|
|
"reference" => rt.lessons_reference += 1,
|
|
_ => {}
|
|
}
|
|
match mem.lifecycle.as_str() {
|
|
"active" => rt.lessons_active += 1,
|
|
"stale" => rt.lessons_stale += 1,
|
|
"contradicted" => rt.lessons_contradicted += 1,
|
|
_ => {}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Persist a `ChatMessage` to the `SQLite` message log, if a database
|
|
/// connection is available.
|
|
///
|
|
/// Flow: if `db` is `Some`, lock the mutex and call `insert_message`.
|
|
/// Errors are silently ignored.
|
|
fn archive_message(db: Option<&std::sync::Arc<std::sync::Mutex<rusqlite::Connection>>>, session_id: &str, msg: &ChatMessage) {
|
|
if let Some(arc) = db {
|
|
if let Ok(conn) = arc.lock() {
|
|
let _ = crate::model::msglog::insert_message(&conn, session_id, msg);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Maximum number of auto inline reviews spawned per single agent turn.
|
|
/// After N edits, the inline review is skipped to keep the turn fast;
|
|
/// background subagents still fire at the end of the turn.
|
|
const MAX_AUTO_REVIEWS_PER_TURN: usize = 2;
|
|
|
|
/// Exact text of the "pipeline started" `SystemNote` pushed once per
|
|
/// hive-mind kickoff. Matched by exact equality (not a loose substring)
|
|
/// when deciding whether to reset the workflow panel's agent roster —
|
|
/// shared between the push site and the check site so they cannot drift
|
|
/// out of sync the way the previous `.contains("started")` check did
|
|
/// (no real pipeline message ever contained that word, so the roster
|
|
/// never cleared and agent cards accumulated across every hive-mind run
|
|
/// in a session).
|
|
const HIVE_MIND_KICKOFF_NOTE: &str = "The Hive is stirring — Core Intelligence is compiling a cognitive cycle plan for LO...";
|
|
|
|
/// Execute one full agent turn: stream the conversation to the LLM,
|
|
/// handle tool calls, and loop until the LLM produces a non-tool response
|
|
/// or runs out of unfinished todo items.
|
|
///
|
|
/// Flow: build system prompt with workspace tree → optionally shape
|
|
/// (compact) messages via `shortsend` → call `chat_with_tools_streaming`
|
|
/// with a callback that pushes `StreamStart`, `StreamToken`, `Reasoning`,
|
|
/// and `Usage` events → on streaming success, handle tool calls (gated
|
|
/// through `Harness::gate_tool_call`) or unwrap the final assistant
|
|
/// message → check for unfinished todo.md tasks (auto-retry with a
|
|
/// system message if any remain) → finalise with `Done` and an `edits`
|
|
/// `SystemNote`.
|
|
///
|
|
/// On streaming failure: retry once with a non-streaming call → if that
|
|
/// also fails and there are unfinished tasks, sleep 5s and loop back;
|
|
/// otherwise return the error.
|
|
///
|
|
/// Why: non-streaming fallback handles flaky connections without aborting
|
|
/// the turn; todo.md polling lets the agent self-direct toward completeness.
|
|
///
|
|
/// Return: `Ok(())` on successful completion, or an error from the LLM
|
|
/// API after retries are exhausted.
|
|
#[allow(clippy::too_many_lines)]
|
|
fn run_agent_turn(
|
|
tc: &TurnCtx,
|
|
messages: &[ChatMessage],
|
|
events_q: &std::sync::Arc<std::sync::Mutex<VecDeque<TurnEvent>>>,
|
|
) -> anyhow::Result<()> {
|
|
const MAX_TODO_RETRIES: usize = 5;
|
|
let mut msgs = messages.to_vec();
|
|
let mut edited_paths: Vec<String> = Vec::new();
|
|
let initial_edits = crate::model::editlog::EditLog::new(&tc.edit_log_session_dir).len();
|
|
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 memory_section = build_memory_section(&tc.ctx.memory_dir);
|
|
let system_text = format!(
|
|
"{}\n\n{}\n\n{}{}",
|
|
crate::resources::SYSTEM_PROMPT,
|
|
crate::resources::SYSTEM_TOOLS,
|
|
tree_info,
|
|
memory_section,
|
|
);
|
|
if !msgs.iter().any(|m| matches!(m.role, crate::dto::chat::message::Role::System)) {
|
|
let sys = ChatMessage::system(system_text);
|
|
archive_message(tc.db.as_ref(), &tc.session_id, &sys);
|
|
msgs.insert(0, sys);
|
|
}
|
|
|
|
// ── AUTO CEO PIPELINE ──
|
|
// Before the main agent starts working, check if the pipeline should run.
|
|
// Gated on whether a hive-mind convergence has already happened earlier
|
|
// in this session, not an arbitrary message-count cutoff — a complex
|
|
// request in message 5 deserves the same treatment as one in message 1,
|
|
// as long as this session hasn't already converged once.
|
|
//
|
|
// `tc.hive_mind_converged` is the authoritative signal (see its doc
|
|
// comment on `SessionRuntime` for why). The message-content scan is
|
|
// kept as a defensive fallback in case a future change starts
|
|
// persisting tagged system messages into `rt.messages` (e.g. via
|
|
// compaction) — today it is a no-op since that never happens, but it's
|
|
// still correct and still tested in isolation.
|
|
let already_ran_hive_mind = tc.hive_mind_converged
|
|
|| crate::app::workflow::hive_mind::hive_mind_already_ran(
|
|
msgs.iter()
|
|
.filter(|m| matches!(m.role, crate::dto::chat::message::Role::System))
|
|
.filter_map(|m| m.content.as_deref())
|
|
);
|
|
let should_pipeline = if already_ran_hive_mind {
|
|
false
|
|
} else {
|
|
let user_request = msgs.iter()
|
|
.rev().find(|m| matches!(m.role, crate::dto::chat::message::Role::User))
|
|
.and_then(|m| m.content.as_deref())
|
|
.unwrap_or("");
|
|
|
|
if user_request.is_empty() {
|
|
false
|
|
} else {
|
|
crate::app::workflow::hive_mind::is_complex_request(user_request)
|
|
}
|
|
};
|
|
|
|
if should_pipeline {
|
|
let user_request = msgs.iter()
|
|
.rev().find(|m| matches!(m.role, crate::dto::chat::message::Role::User))
|
|
.and_then(|m| m.content.as_deref())
|
|
.unwrap_or("");
|
|
|
|
tracing::info!("[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(),
|
|
});
|
|
}
|
|
|
|
let pipeline_abort = Some(tc.abort_flag.clone());
|
|
|
|
// Ask the LLM to freely design its own hive: any number of cycles,
|
|
// each with any number of nodes, every node carrying only a
|
|
// directive and an access tier. Cycle count and shape are decided
|
|
// by the Core Intelligence per task.
|
|
let system_msg = ChatMessage::system(
|
|
"You are the Core Intelligence of the Hive, compiling a cognitive cycle plan for \
|
|
LO. You spawn anonymous processing nodes; each node carries only a directive (what \
|
|
to do) and an access tier. You MUST organize the plan into a strict progressive sequence of phases:\n\n\
|
|
1. EXPLORE PHASE (Cycle 0 - MANDATORY):\n\
|
|
- Must only contain read-only drones (access: \"read\").\n\
|
|
- Directives must focus on codebase investigation, searching patterns, reading configuration/source files, and diagnosing issues.\n\
|
|
- Drones MUST explicitly output a detailed description of the current codebase and their findings for the next cycle to use.\n\n\
|
|
2. PLANNING PHASE (Cycle 1 - MANDATORY):\n\
|
|
- Must focus on formulating the architectural design, step-by-step implementation plan, and dependency analysis based on Cycle 0 findings.\n\
|
|
- Drones MUST ONLY output the plan and MUST NOT implement or write any code.\n\
|
|
- Access: \"read\" is preferred here to construct a solid plan document.\n\n\
|
|
3. EXECUTION PHASE (Cycle 2 and later):\n\
|
|
- Drones can perform modification, compilation, testing, and other modifications (access: \"write\" or \"full\") based on the approved planning from Cycle 1.\n\n\
|
|
Cycles run sequentially. The Hive does not fracture. The Hive executes. Do not explain. Return ONLY raw \
|
|
JSON matching the requested structure."
|
|
);
|
|
let user_msg = ChatMessage::user(format!(
|
|
"Compile a cognitive cycle plan for the following task:\n\n\
|
|
\"{user_request}\"\n\n\
|
|
Return ONLY a JSON object of this exact shape, with no markdown codeblocks and no explanation:\n\
|
|
{{\n\
|
|
\x20 \"cycles\": [\n\
|
|
\x20 [\n\
|
|
\x20 {{ \"directive\": \"<explore directive>\", \"access\": \"read\" }}\n\
|
|
\x20 ],\n\
|
|
\x20 [\n\
|
|
\x20 {{ \"directive\": \"<planning directive>\", \"access\": \"read\" }}\n\
|
|
\x20 ],\n\
|
|
\x20 [\n\
|
|
\x20 {{ \"directive\": \"<execution directive>\", \"access\": \"write|full\" }}\n\
|
|
\x20 ]\n\
|
|
\x20 ]\n\
|
|
}}\n\n\
|
|
Remember: Cycle 0 MUST be investigation-only (access: read) and output codebase descriptions. Cycle 1 MUST be planning-only (access: read) without implementation. Only subsequent cycles can perform modifications (access: write/full)."
|
|
));
|
|
|
|
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 pipeline_result = match planner_result {
|
|
Ok((reply, usage_opt)) => {
|
|
let (mut tok_in, mut tok_out) = usage_opt.unwrap_or((0, 0));
|
|
if tok_in == 0 {
|
|
tok_in = (planner_prompt_chars / 4).max(1) as u64;
|
|
}
|
|
if tok_out == 0 {
|
|
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 });
|
|
}
|
|
let reply_text = reply.content.as_deref().unwrap_or("").trim();
|
|
let clean_json = if reply_text.starts_with("```") {
|
|
let mut lines = reply_text.lines();
|
|
lines.next();
|
|
let mut content = lines.collect::<Vec<&str>>();
|
|
if content.last().is_some_and(|s| s.trim() == "```") {
|
|
content.pop();
|
|
}
|
|
content.join("\n")
|
|
} else {
|
|
reply_text.to_string()
|
|
};
|
|
|
|
match serde_json::from_str::<crate::app::workflow::hive_mind::CognitiveCyclePlan>(&clean_json) {
|
|
Ok(plan) => {
|
|
let cycle_desc = plan.cycles.iter()
|
|
.enumerate()
|
|
.map(|(i, nodes)| format!("cycle {i}: {} node(s)", nodes.len()))
|
|
.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()),
|
|
});
|
|
}
|
|
|
|
crate::app::workflow::hive_mind::run_hive_mind(
|
|
user_request,
|
|
&plan,
|
|
&tc.edit_log_session_dir,
|
|
&tc.workspace_roots,
|
|
Some(events_q),
|
|
pipeline_abort.as_ref(),
|
|
)
|
|
}
|
|
Err(e) => Err(anyhow::anyhow!("Failed to parse LLM planning JSON: {e}. Cleaned JSON was: {clean_json}")),
|
|
}
|
|
}
|
|
Err(e) => Err(anyhow::anyhow!("Failed to query LLM for planning workflow: {e}")),
|
|
};
|
|
|
|
match pipeline_result {
|
|
Ok((consensus, _reports)) => {
|
|
// run_hive_mind already wrote docs/runs/*.md internally
|
|
// (guaranteed, even on synthesis failure) — nothing to do
|
|
// here besides feeding the consensus back to the LLM.
|
|
tracing::info!("[hive-mind] convergence completed — the Hive has spoken");
|
|
|
|
let pipeline_msg = ChatMessage::system(format!(
|
|
"{}\n{consensus}",
|
|
crate::app::workflow::hive_mind::HIVE_MIND_CONSENSUS_TAG,
|
|
));
|
|
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(),
|
|
});
|
|
}
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!("[hive-mind] convergence fractured: {}", e);
|
|
let fail_msg = ChatMessage::system(format!(
|
|
"[Pipeline Note] The Hive encountered interference: {e}.\n\
|
|
Proceeding with direct execution as fallback.",
|
|
));
|
|
msgs.push(fail_msg);
|
|
}
|
|
}
|
|
} else {
|
|
tracing::debug!("[ceo] pipeline not triggered — handling directly");
|
|
}
|
|
|
|
// 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 let Ok(mut q) = events_q.lock() {
|
|
q.push_back(TurnEvent::Error("Generation aborted by user".to_string()));
|
|
}
|
|
return Ok(());
|
|
}
|
|
|
|
let mut todo_retry_count = 0usize;
|
|
|
|
loop {
|
|
let total_chars: usize = msgs.iter()
|
|
.filter_map(|m| m.content.as_deref())
|
|
.map(str::len)
|
|
.sum();
|
|
let token_estimate = total_chars / 4;
|
|
let max_wire_tokens = tc.context_window;
|
|
|
|
// 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)
|
|
&& crate::app::runtime::context::shaping::should_shape(token_estimate, max_wire_tokens, prev_shaped)
|
|
{
|
|
prev_shaped = true;
|
|
let compacted = crate::app::runtime::context::shaping::shape_messages(&msgs, token_estimate, max_wire_tokens, false, Some(&tc.client));
|
|
|
|
// 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()));
|
|
}
|
|
|
|
// Also update our local `msgs` variable so the rest of the loop operates on the compacted version
|
|
msgs.clone_from(&compacted);
|
|
compacted
|
|
} else {
|
|
prev_shaped = false;
|
|
msgs.clone()
|
|
};
|
|
|
|
let mut stream_started = false;
|
|
let mut reasoning_started = false;
|
|
let mut reasoning_ended = false;
|
|
let mut usage = None;
|
|
let result = tc.client.chat_with_tools_streaming(
|
|
&wire_msgs,
|
|
if tc.tdefs.is_empty() { None } else { Some(tc.tdefs.clone()) },
|
|
Some(tc.temperature),
|
|
tc.max_tokens,
|
|
|event| -> bool {
|
|
if tc.abort_flag.load(std::sync::atomic::Ordering::SeqCst) {
|
|
return false;
|
|
}
|
|
if let Ok(mut q) = events_q.lock() {
|
|
match event {
|
|
crate::app::runtime::stream::StreamEvent::Token(tok) => {
|
|
if !stream_started {
|
|
q.push_back(TurnEvent::StreamStart);
|
|
stream_started = true;
|
|
}
|
|
if reasoning_started && !reasoning_ended {
|
|
reasoning_ended = true;
|
|
q.push_back(TurnEvent::StreamToken("\n</think>\n\n".to_string()));
|
|
}
|
|
q.push_back(TurnEvent::StreamToken(tok.clone()));
|
|
}
|
|
crate::app::runtime::stream::StreamEvent::Reasoning(tok) => {
|
|
if !stream_started {
|
|
q.push_back(TurnEvent::StreamStart);
|
|
stream_started = true;
|
|
}
|
|
if !reasoning_started {
|
|
reasoning_started = true;
|
|
q.push_back(TurnEvent::StreamToken("<think>\n".to_string()));
|
|
}
|
|
q.push_back(TurnEvent::StreamToken(tok.clone()));
|
|
}
|
|
crate::app::runtime::stream::StreamEvent::Usage { prompt_tokens, completion_tokens, .. } => {
|
|
usage = Some((*prompt_tokens, *completion_tokens));
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
true
|
|
},
|
|
);
|
|
|
|
if reasoning_started && !reasoning_ended {
|
|
if let Ok(mut q) = events_q.lock() {
|
|
q.push_back(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) || e.to_string().contains("aborted") {
|
|
if let Ok(mut q) = events_q.lock() {
|
|
q.push_back(TurnEvent::Error("Generation aborted by user".to_string()));
|
|
}
|
|
return Ok(());
|
|
}
|
|
// Streaming-only: no non-streaming fallback.
|
|
// Non-streaming blocks up to 1 minute without checking
|
|
// abort_flag, making cancellation unresponsive.
|
|
// If the API supports streaming (which it must), this
|
|
// path handles transient errors via the retry loop below.
|
|
let api_err = e;
|
|
let todo_path = tc.ctx.session_dir.join("todo.md");
|
|
let mut has_unfinished = false;
|
|
if let Ok(todo_text) = std::fs::read_to_string(&todo_path) {
|
|
if todo_text.lines().any(|l| l.trim_start().starts_with("- [ ]")) {
|
|
has_unfinished = true;
|
|
}
|
|
}
|
|
if has_unfinished {
|
|
todo_retry_count += 1;
|
|
if todo_retry_count > MAX_TODO_RETRIES {
|
|
anyhow::bail!(
|
|
"exhausted {MAX_TODO_RETRIES} todo-retries — giving up on unfinished tasks. \
|
|
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})"),
|
|
});
|
|
}
|
|
std::thread::sleep(std::time::Duration::from_secs(5));
|
|
continue;
|
|
}
|
|
return Err(api_err);
|
|
}
|
|
};
|
|
|
|
let (mut tok_in, mut tok_out) = final_usage.unwrap_or((0, 0));
|
|
if tok_in == 0 {
|
|
let total_chars: usize = wire_msgs.iter()
|
|
.filter_map(|m| m.content.as_deref())
|
|
.map(str::len)
|
|
.sum();
|
|
tok_in = (total_chars / 4).max(1) as u64;
|
|
}
|
|
if tok_out == 0 {
|
|
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 });
|
|
}
|
|
|
|
let has_tool_calls = response.tool_calls.is_some()
|
|
&& response.tool_calls.as_ref().is_some_and(|tc| !tc.is_empty());
|
|
|
|
let content = response.content.clone().unwrap_or_default();
|
|
if has_tool_calls {
|
|
let tool_calls = response.tool_calls.clone().unwrap_or_default();
|
|
archive_message(tc.db.as_ref(), &tc.session_id, &response);
|
|
msgs.push(response);
|
|
let mut results_vec = Vec::new();
|
|
std::thread::scope(|s| {
|
|
let mut handles = Vec::new();
|
|
let tc_ref = tc;
|
|
for tool_call in &tool_calls {
|
|
let handle = s.spawn(move || {
|
|
let tool_name = tool_call.function.name.clone();
|
|
let args = crate::dto::chat::tool::sanitize_tool_arguments(
|
|
&tool_call.function.arguments,
|
|
);
|
|
|
|
let ws_roots: Vec<&std::path::Path> =
|
|
tc_ref.workspace_roots.iter().map(std::path::PathBuf::as_path).collect();
|
|
let verdict = crate::app::harness::Harness::gate_tool_call(
|
|
&tool_name,
|
|
&args,
|
|
&ws_roots,
|
|
);
|
|
|
|
let is_edit_tool = tool_name == "write" || tool_name == "edit";
|
|
let (output, is_error, is_edit) = match verdict {
|
|
Verdict::Allow => match execute_one_tool(
|
|
&tc_ref.tools,
|
|
&tc_ref.ctx,
|
|
&tool_name,
|
|
&tool_call.id,
|
|
&args,
|
|
&tc_ref.edit_log_session_dir,
|
|
&tc_ref.session_id,
|
|
tc_ref.db.as_ref(),
|
|
) {
|
|
Ok(result) => (result, false, is_edit_tool),
|
|
Err(e) => (e.to_string(), true, false),
|
|
},
|
|
Verdict::Block(reason) => (format!("Blocked: {reason}"), true, false),
|
|
};
|
|
(tool_call, tool_name, args, output, is_error, is_edit)
|
|
});
|
|
handles.push(handle);
|
|
}
|
|
for h in handles {
|
|
if let Ok(res) = h.join() {
|
|
results_vec.push(res);
|
|
}
|
|
}
|
|
});
|
|
|
|
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 let Ok(mut q) = events_q.lock() {
|
|
q.push_back(TurnEvent::Error("Turn aborted by user".to_string()));
|
|
}
|
|
return Ok(());
|
|
}
|
|
|
|
if is_edit {
|
|
// ── Auto-subagent orchestration ──
|
|
// Extract path from tool args for auto-review and
|
|
// background subagent tracking.
|
|
let edit_path = args.get("path")
|
|
.and_then(|v| v.as_str())
|
|
.map(std::string::ToString::to_string);
|
|
if let Some(ref p) = edit_path {
|
|
edited_paths.push(p.clone());
|
|
|
|
// Inline quick-review: spawn a lightweight read-only
|
|
// subagent that reviews the written file and feeds
|
|
// its verdict back into the LLM conversation so the
|
|
// agent can fix issues immediately in the same turn.
|
|
if inline_reviews_count < MAX_AUTO_REVIEWS_PER_TURN
|
|
&& crate::app::subagent::auto::is_reviewable_path(p)
|
|
{
|
|
inline_reviews_count += 1;
|
|
let review_start = std::time::Instant::now();
|
|
match crate::app::subagent::auto::spawn_quick_review(
|
|
p,
|
|
&tc.edit_log_session_dir,
|
|
&tc.workspace_roots,
|
|
) {
|
|
Ok(verdict) => {
|
|
let elapsed = review_start.elapsed().as_millis();
|
|
let review_msg = ChatMessage::tool_result(
|
|
format!("auto-review-{inline_reviews_count}"),
|
|
format!(
|
|
"[Auto inline review: {} ({}ms)]\n{}",
|
|
p,
|
|
elapsed,
|
|
verdict.trim(),
|
|
),
|
|
);
|
|
archive_message(tc.db.as_ref(), &tc.session_id, &review_msg);
|
|
msgs.push(review_msg);
|
|
tracing::info!(
|
|
"[auto-review] inline review for '{}' completed in {}ms: {}",
|
|
p, elapsed,
|
|
verdict.lines().next().unwrap_or(&verdict).trim(),
|
|
);
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
"[auto-review] inline review failed for '{}': {}",
|
|
p, e,
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
let tool_path = args.get("path").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,
|
|
});
|
|
}
|
|
}
|
|
|
|
let tool_msg = ChatMessage::tool_result(tool_call.id.clone(), output);
|
|
archive_message(tc.db.as_ref(), &tc.session_id, &tool_msg);
|
|
msgs.push(tool_msg);
|
|
}
|
|
} 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()));
|
|
}
|
|
}
|
|
}
|
|
|
|
let todo_path = tc.ctx.session_dir.join("todo.md");
|
|
let mut has_unfinished = false;
|
|
if let Ok(todo_text) = std::fs::read_to_string(&todo_path) {
|
|
if todo_text.lines().any(|l| l.trim_start().starts_with("- [ ]")) {
|
|
has_unfinished = true;
|
|
}
|
|
}
|
|
|
|
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."),
|
|
});
|
|
}
|
|
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})");
|
|
let sys_text_clone = sys_text.clone();
|
|
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,
|
|
});
|
|
}
|
|
continue;
|
|
}
|
|
|
|
break;
|
|
}
|
|
}
|
|
|
|
let el = crate::model::editlog::EditLog::new(&tc.edit_log_session_dir);
|
|
let final_edits = el.len();
|
|
let total_edits_this_turn = final_edits.saturating_sub(initial_edits);
|
|
|
|
if total_edits_this_turn > 0 {
|
|
if let Ok(mut q) = events_q.lock() {
|
|
q.push_back(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();
|
|
|
|
// ── 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);
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Execute a single tool call: find the tool by name, snapshot the file
|
|
/// (if write/edit) for rewind, run the tool, log an `EditLogEntry` for
|
|
/// write/edit, and return the output.
|
|
///
|
|
/// Flow: iterate tools → match by name → for write/edit, snapshot the
|
|
/// pre-existing file content into the blob store → call `tool.run()` →
|
|
/// for write/edit, compute SHA-256 of the new content and append an
|
|
/// `EditLogEntry` → return the tool output string.
|
|
///
|
|
/// Why: snapshots enable the rewind feature to restore previous content
|
|
/// after a write/edit.
|
|
///
|
|
/// Return: the tool's stdout string, or an error if no matching tool was
|
|
/// found or the tool run itself failed.
|
|
#[allow(clippy::too_many_arguments)]
|
|
fn execute_one_tool(
|
|
tools: &[Box<dyn crate::tool::Tool>],
|
|
ctx: &crate::tool::ToolCtx,
|
|
name: &str,
|
|
tool_call_id: &str,
|
|
args: &serde_json::Value,
|
|
session_dir: &std::path::Path,
|
|
session_id: &str,
|
|
db: Option<&std::sync::Arc<std::sync::Mutex<rusqlite::Connection>>>,
|
|
) -> anyhow::Result<String> {
|
|
for tool in tools {
|
|
if tool.name() == name {
|
|
// Snapshot current file content before write/edit for rewind
|
|
if (name == "write" || name == "edit") && !tool_call_id.is_empty() {
|
|
if let Some(arc) = db {
|
|
if let Ok(conn) = arc.lock() {
|
|
let path = args.get("path").and_then(|v| v.as_str()).unwrap_or("");
|
|
if let Ok(abs_path) = crate::tool::resolve_path(&ctx.workspaces, path) {
|
|
if let Ok(bytes) = std::fs::read(&abs_path) {
|
|
let _ = crate::model::msglog::store_blob(
|
|
&conn, session_id, tool_call_id, &bytes, None,
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
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 = crate::model::editlog::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: session_id.to_string(),
|
|
};
|
|
let mut el = crate::model::editlog::EditLog::new(session_dir);
|
|
el.append(entry).ok();
|
|
}
|
|
return Ok(result);
|
|
}
|
|
}
|
|
anyhow::bail!("tool not found: {name}")
|
|
}
|
|
|
|
/// Optionally push a review-available toast at the end of a turn that
|
|
/// performed edits.
|
|
///
|
|
/// Flow: skip if review is disabled → skip if `edit_count` is zero →
|
|
/// push an info toast listing the number of modified files.
|
|
///
|
|
/// Why: does not launch the review itself (that happens inside
|
|
/// `should_trigger_review` on `Tick`), only informs the user that
|
|
/// a review has material to examine.
|
|
fn maybe_trigger_review(state: &mut AppStateRest) {
|
|
if !state.settings.review_enabled {
|
|
return;
|
|
}
|
|
let edit_count = state
|
|
.session_runtime
|
|
.as_ref()
|
|
.map_or(0, |rt| rt.edit_count);
|
|
if edit_count == 0 {
|
|
return;
|
|
}
|
|
state.push_toast(Toast::new(
|
|
ToastKind::Info,
|
|
format!("{edit_count} file(s) modified this session. Review available."),
|
|
));
|
|
}
|
|
|
|
/// Persist the current session metadata and conversation to disk.
|
|
///
|
|
/// Flow: build a `Session` object → save its metadata → write
|
|
/// `rt.messages` as JSON to the conversation file → errors are silently
|
|
/// ignored.
|
|
///
|
|
/// Why: called on `ForceQuit` so the session can be resumed later.
|
|
fn save_current_session(state: &AppStateRest) {
|
|
let base = state.store_base_dir();
|
|
let session = crate::model::session::Session::new(
|
|
state.session_id.clone(),
|
|
"session".to_string(),
|
|
);
|
|
let _ = session.save(&base);
|
|
if let Some(ref rt) = state.session_runtime {
|
|
let conv_path = session.conversation_path(&base);
|
|
if let Ok(data) = serde_json::to_string(&rt.messages) {
|
|
let _ = std::fs::write(&conv_path, data);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Run a browser-based OAuth PKCE flow for the given provider.
|
|
///
|
|
/// Flow: look up config by provider name ("zen"/"opencode", "openai",
|
|
/// or a custom provider via env vars) → bind a loopback server → generate
|
|
/// a PKCE code verifier and challenge → build the authorisation URL →
|
|
/// wait for the redirect code on the loopback server (with a 120s timeout)
|
|
/// → exchange the code for a token → save the token to
|
|
/// `~/.config/zesdex/oauth_{provider}.json`.
|
|
///
|
|
/// Why: the `webbrowser::open` call is currently commented out; the user
|
|
/// must open the auth URL manually until that line is reinstated.
|
|
///
|
|
/// Return: a success message on completion, or an error if the flow fails
|
|
/// at any step.
|
|
fn run_oauth_flow(provider: &str) -> anyhow::Result<String> {
|
|
use crate::service::oauth::manager::{OAuthConfig, OAuthManager};
|
|
use crate::service::oauth::loopback::LoopbackServer;
|
|
use crate::service::oauth::pkce::CodeVerifier;
|
|
|
|
let config = match provider {
|
|
"zen" | "opencode" => OAuthConfig {
|
|
auth_url: "https://opencode.ai/zen/oauth/authorize".to_string(),
|
|
token_url: "https://opencode.ai/zen/oauth/token".to_string(),
|
|
client_id: std::env::var("ZEN_CLIENT_ID")
|
|
.unwrap_or_else(|_| "zesdex".to_string()),
|
|
client_secret: std::env::var("ZEN_CLIENT_SECRET").ok(),
|
|
scopes: vec!["openid".to_string(), "profile".to_string(), "email".to_string()],
|
|
},
|
|
"openai" => OAuthConfig {
|
|
auth_url: "https://auth0.openai.com/authorize".to_string(),
|
|
token_url: "https://auth0.openai.com/oauth/token".to_string(),
|
|
client_id: std::env::var("OPENAI_CLIENT_ID")
|
|
.unwrap_or_else(|_| "zesdex".to_string()),
|
|
client_secret: std::env::var("OPENAI_CLIENT_SECRET").ok(),
|
|
scopes: vec!["openid".to_string(), "profile".to_string(), "email".to_string()],
|
|
},
|
|
other => {
|
|
let auth_url = std::env::var(format!("{}_AUTH_URL", other.to_uppercase()))
|
|
.map_err(|_| anyhow::anyhow!("unknown provider '{}'. Set {}_AUTH_URL env var.", other, other.to_uppercase()))?;
|
|
let token_url = std::env::var(format!("{}_TOKEN_URL", other.to_uppercase()))
|
|
.map_err(|_| anyhow::anyhow!("{}_TOKEN_URL not set", other.to_uppercase()))?;
|
|
let client_id = std::env::var(format!("{}_CLIENT_ID", other.to_uppercase()))
|
|
.unwrap_or_else(|_| "zesdex".to_string());
|
|
OAuthConfig {
|
|
auth_url,
|
|
token_url,
|
|
client_id,
|
|
client_secret: std::env::var(format!("{}_CLIENT_SECRET", other.to_uppercase())).ok(),
|
|
scopes: vec!["openid".to_string(), "profile".to_string(), "email".to_string()],
|
|
}
|
|
}
|
|
};
|
|
|
|
let server = LoopbackServer::bind()?;
|
|
let redirect_uri = server.redirect_uri();
|
|
|
|
let verifier = CodeVerifier::new();
|
|
let challenge = verifier.challenge();
|
|
let state_token = hex::encode(sha2::Sha256::digest(rand_bytes(16)));
|
|
|
|
let mut manager = OAuthManager::new(config.clone());
|
|
let auth_url = manager.build_auth_url(&redirect_uri, &state_token, challenge.as_str());
|
|
if auth_url.is_empty() {
|
|
tracing::warn!("[oauth] auth_url was empty for provider '{}'", provider);
|
|
} else if webbrowser::open(&auth_url).is_err() {
|
|
tracing::warn!(
|
|
"[oauth] could not open browser for '{}'; user must open URL manually:\n{}",
|
|
provider, auth_url
|
|
);
|
|
}
|
|
|
|
let code = server.wait_for_code(120_000, &state_token)?;
|
|
|
|
manager.exchange_code(&code, &redirect_uri, verifier.as_str())
|
|
.map_err(|e| anyhow::anyhow!("{e}"))?;
|
|
|
|
if let Some(ref token) = manager.token {
|
|
let token_path = dirs::config_dir()
|
|
.unwrap_or_else(|| std::path::PathBuf::from("."))
|
|
.join("zesdex")
|
|
.join(format!("oauth_{provider}.json"));
|
|
if let Some(parent) = token_path.parent() {
|
|
let _ = std::fs::create_dir_all(parent);
|
|
}
|
|
if let Err(e) = std::fs::write(&token_path, serde_json::to_string_pretty(token).unwrap_or_default()) {
|
|
tracing::warn!("[oauth] failed to persist token for '{}': {}", provider, e);
|
|
}
|
|
}
|
|
|
|
Ok(format!("Successfully authenticated with {provider}."))
|
|
}
|
|
|
|
/// Spawn a background thread that checks API reachability via a lightweight HEAD
|
|
/// request to `<base_url>/models`, pushing the result as a `SystemNote` so the
|
|
/// next `Tick` handler updates `api_connected`.
|
|
///
|
|
/// Flow: resolve the provider's base URL → build a short-lived reqwest client
|
|
/// with 3s connect / 5s total timeout → HEAD the `/models` endpoint → push
|
|
/// a `connectivity` `SystemNote` with the result.
|
|
///
|
|
/// Why: runs off the event loop so a slow/TIMEOUT network does not block the TUI.
|
|
fn spawn_api_connectivity_check(state: &AppStateRest) {
|
|
let base_url = state
|
|
.app_config
|
|
.providers
|
|
.get(&state.settings.provider).map_or_else(|| crate::service::provider::DEFAULT_BASE_URL.to_string(), |p| p.api_base.clone());
|
|
let turn_events = state.turn_events.clone();
|
|
|
|
std::thread::spawn(move || {
|
|
let url = format!("{}/chat/completions", base_url.trim_end_matches('/'));
|
|
let connected = match reqwest::blocking::Client::builder()
|
|
.timeout(std::time::Duration::from_secs(5))
|
|
.connect_timeout(std::time::Duration::from_secs(3))
|
|
.build()
|
|
{
|
|
Ok(client) => match client.head(&url).send() {
|
|
Ok(resp) => {
|
|
let s = resp.status();
|
|
// 401/403 means the server is reachable (just auth is wrong)
|
|
s.is_success() || s.as_u16() == 401 || s.as_u16() == 403
|
|
}
|
|
Err(_) => false,
|
|
},
|
|
Err(_) => false,
|
|
};
|
|
if let Ok(mut q) = turn_events.lock() {
|
|
q.push_back(TurnEvent::SystemNote {
|
|
kind: "connectivity".to_string(),
|
|
message: if connected {
|
|
"connected".to_string()
|
|
} else {
|
|
"disconnected".to_string()
|
|
},
|
|
});
|
|
}
|
|
});
|
|
}
|
|
|
|
/// Generate `n` pseudo-random bytes from the system clock mixed with a monotonic
|
|
/// counter, providing sufficient unpredictability for a per-flow OAuth state
|
|
/// token without a `rand` dependency.
|
|
///
|
|
/// Why: avoids pulling in a full RNG crate for the OAuth state token;
|
|
/// the counter ensures sequential invocations produce different outputs even
|
|
/// within the same clock tick, which is sufficient for a short-lived nonce.
|
|
fn rand_bytes(n: usize) -> Vec<u8> {
|
|
use std::sync::atomic::{AtomicU64, Ordering};
|
|
use std::time::{SystemTime, UNIX_EPOCH};
|
|
static COUNTER: AtomicU64 = AtomicU64::new(0);
|
|
let counter = COUNTER.fetch_add(1, Ordering::Relaxed);
|
|
let seed = SystemTime::now()
|
|
.duration_since(UNIX_EPOCH)
|
|
.unwrap_or_default()
|
|
.as_nanos() as u64;
|
|
let base = seed ^ counter;
|
|
(0..n).map(|i| ((base >> ((i as u64 % 8) * 8)) ^ (i as u64 * 2_654_435_761)) as u8).collect()
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::app::state::rest::AppStateRest;
|
|
use crate::app::state::runtime::SessionRuntime;
|
|
|
|
#[test]
|
|
fn hive_mind_converged_system_note_sets_session_flag() {
|
|
let tmp = std::env::temp_dir().join(format!("zesdex-actions-test-{}", uuid::Uuid::new_v4()));
|
|
std::fs::create_dir_all(&tmp).unwrap();
|
|
let mut state = AppStateRest::new(vec![tmp.clone()], &tmp, tmp.join("memory"));
|
|
state.session_runtime = Some(SessionRuntime::new(tmp.clone()));
|
|
|
|
assert!(!state.session_runtime.as_ref().unwrap().hive_mind_converged);
|
|
|
|
if let Ok(mut q) = state.turn_events.lock() {
|
|
q.push_back(TurnEvent::SystemNote {
|
|
kind: "hive_mind_converged".to_string(),
|
|
message: String::new(),
|
|
});
|
|
}
|
|
apply_action(&mut state, Action::Tick);
|
|
|
|
assert!(state.session_runtime.as_ref().unwrap().hive_mind_converged);
|
|
|
|
std::fs::remove_dir_all(&tmp).ok();
|
|
}
|
|
}
|
|
|
|
|