admin/openhuman
1
0
Fork 0
mirror of https://github.com/tinyhumansai/openhuman.git synced 2026-07-27 21:08:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix false service gate blocking when service is already running (#64)

Browse Source 
* Update CLAUDE.md to clarify pull request workflow and enhance documentation

- Added a note to emphasize the importance of opening pull requests against the upstream repository instead of a fork's default remote, ensuring better collaboration and code management.
- Improved the overall clarity of the Git workflow section, contributing to a more streamlined development process.

* Refactor REPL session handling and remove deprecated chat methods

- Removed the `repl_session_chat` functionality from the schema and related files, streamlining the REPL session management.
- Introduced a new `start_chat` method in the web channel provider for handling chat interactions, enhancing the chat system's architecture.
- Updated socket handling to improve event emission with alias support, ensuring better communication across the application.
- Enhanced error handling and logging for chat operations, providing clearer feedback during interactions.
- Consolidated chat-related logic into the new web channel module, improving maintainability and organization.

* Refactor apiClient to improve token management and eliminate circular dependencies

- Renamed `_getToken` to `authTokenGetterRef` for clarity and to avoid clashing with transpiled private method names.
- Updated the `setStoreForApiClient` function to use the new `authTokenGetterRef` variable, ensuring the `apiClient` remains free of direct imports from `store/index`.
- Enhanced the `ApiClient` class by renaming the `getToken` method to `resolveAuthToken`, improving readability and consistency in token retrieval logic.
- Adjusted comments in `apiClient.ts` and `store/index.ts` to reflect the changes and clarify the purpose of the lazy store accessor.

* Enhance API client integration with store for improved token management

- Introduced `setStoreForApiClient` in `main.tsx` to bind the Redux store's auth token for API requests, ensuring seamless token access.
- Updated comments in `apiClient.ts` to clarify the lazy token accessor's purpose and prevent circular dependencies.
- Removed the previous direct call to `setStoreForApiClient` from `store/index.ts`, centralizing the setup in `main.tsx` for better organization.
- Enhanced test setup to include the new store binding, ensuring consistent token retrieval during testing.

* Refactor web channel event handling and enhance chat functionality

- Updated `WebChatSseEvent` structure to include new fields: `tool_name`, `skill_id`, `args`, `output`, `success`, and `round`, improving the granularity of event data.
- Refactored event emission logic in `emit_web_chat_event` to utilize the new fields, ensuring more informative payloads for tool calls and results.
- Enhanced the `start_chat` and `cancel_chat` functions to accommodate the new event structure, improving chat session management.
- Introduced a new function `publish_tool_events_from_history` to streamline the emission of tool call and result events from chat history, enhancing the overall chat experience.

* Refactor web channel integration and enhance controller management

- Updated references from the deprecated `web_channel` module to the new `channels::providers::web` structure, improving code organization and maintainability.
- Refactored the `build_registered_controllers` and `build_declared_controller_schemas` functions to utilize the new web channel controller methods, ensuring consistency across the application.
- Removed the obsolete `web_channel` module and its associated files, streamlining the codebase and enhancing clarity in the channel management system.
- Introduced new functions for managing web channel events and schemas, improving the overall architecture of the web channel integration.

* Refactor main.tsx and test setup for improved organization

- Reordered imports in `main.tsx` to enhance clarity and maintainability, ensuring that `setStoreForApiClient` is called with the correct store reference.
- Removed unnecessary whitespace in `setup.ts`, streamlining the test setup file for better readability.

* Fix service gate false blocking when service is running

* Support soft-pass daemon gate and harden macOS service detection

* Extend list-files fallback trigger phrases in agent loop

* Replace list-files fallback with tool-call repair retry

* Log full system prompt and drop tool-call repair flow
This commit is contained in:
Steven Enamakel
2026-03-29 21:08:38 -07:00
committed by GitHub
parent 7ad3c73ac3
commit 22435ed631
24 changed files with 846 additions and 688 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CLAUDE.md
+1
View File
@@ -270,6 +270,7 @@ Skills runtime uses **QuickJS** (`rquickjs`) in **`src/openhuman/skills/`** (e.g
## Git workflow
- **Open pull requests on upstream** — Always create PRs against **[tinyhumansai/openhuman](https://github.com/tinyhumansai/openhuman)** ([pull requests](https://github.com/tinyhumansai/openhuman/pulls)), not only a fork’s default remote, unless the workflow explicitly says otherwise.
- **Public repo**; push to your working branch; PRs target **`main`**.
- Use [`.github/pull_request_template.md`](.github/pull_request_template.md); AI-generated PR text should follow its sections and checklist.
app/schema.json
-26
View File
@@ -136,19 +136,6 @@
{ "comment": "Agent response payload.", "name": "response", "required": true, "ty": "Json" }
]
},
{
"description": "Send a message through REPL agent session.",
"function": "repl_session_chat",
"inputs": [
{ "comment": "REPL session id.", "name": "session_id", "required": true, "ty": "String" },
{ "comment": "User message.", "name": "message", "required": true, "ty": "String" }
],
"method": "openhuman.agent_repl_session_chat",
"namespace": "agent",
"outputs": [
{ "comment": "Session chat response.", "name": "response", "required": true, "ty": "Json" }
]
},
{
"description": "Terminate REPL session.",
"function": "repl_session_end",
@@ -1285,19 +1272,6 @@
{ "comment": "Agent response payload.", "name": "response", "required": true, "ty": "Json" }
]
},
{
"description": "Send a message through REPL agent session.",
"function": "agent_repl_session_chat",
"inputs": [
{ "comment": "REPL session id.", "name": "session_id", "required": true, "ty": "String" },
{ "comment": "User message.", "name": "message", "required": true, "ty": "String" }
],
"method": "openhuman.local_ai_agent_repl_session_chat",
"namespace": "local_ai",
"outputs": [
{ "comment": "Session chat response.", "name": "response", "required": true, "ty": "Json" }
]
},
{
"description": "Terminate REPL session.",
"function": "agent_repl_session_end",
app/src-tauri/src/commands/openhuman.rs
+15 -9
View File
@@ -104,6 +104,12 @@ struct WebChatSseEvent {
full_response: Option<String>,
message: Option<String>,
error_type: Option<String>,
tool_name: Option<String>,
skill_id: Option<String>,
args: Option<Value>,
output: Option<String>,
success: Option<bool>,
round: Option<u32>,
}
fn core_events_url() -> String {
@@ -168,10 +174,10 @@ fn emit_web_chat_event(app: &AppHandle, event: WebChatSseEvent) {
"tool_call" => {
let payload = serde_json::json!({
"thread_id": event.thread_id,
"tool_name": "unknown",
"skill_id": "web_channel",
"args": {},
"round": 0,
"tool_name": event.tool_name.unwrap_or_else(|| "unknown".to_string()),
"skill_id": event.skill_id.unwrap_or_else(|| "web_channel".to_string()),
"args": event.args.unwrap_or_else(|| serde_json::json!({})),
"round": event.round.unwrap_or(0),
"request_id": event.request_id,
});
let _ = app.emit("chat:tool_call", payload);
@@ -179,11 +185,11 @@ fn emit_web_chat_event(app: &AppHandle, event: WebChatSseEvent) {
"tool_result" => {
let payload = serde_json::json!({
"thread_id": event.thread_id,
"tool_name": "unknown",
"skill_id": "web_channel",
"output": "",
"success": true,
"round": 0,
"tool_name": event.tool_name.unwrap_or_else(|| "unknown".to_string()),
"skill_id": event.skill_id.unwrap_or_else(|| "web_channel".to_string()),
"output": event.output.unwrap_or_default(),
"success": event.success.unwrap_or(true),
"round": event.round.unwrap_or(0),
"request_id": event.request_id,
});
let _ = app.emit("chat:tool_result", payload);
app/src/components/daemon/ServiceBlockingGate.tsx
+19 -7
View File
@@ -22,6 +22,9 @@ type RefreshOptions = { showChecking?: boolean; clearError?: boolean };
const normalizeServiceState = (state: ServiceState | undefined): string => {
if (!state) return 'Unknown';
if (typeof state === 'string') return state;
if ('Running' in state) return 'Running';
if ('Stopped' in state) return 'Stopped';
if ('NotInstalled' in state) return 'NotInstalled';
if ('Unknown' in state) return `Unknown(${state.Unknown})`;
return 'Unknown';
};
@@ -50,26 +53,35 @@ const ServiceBlockingGate = ({ children }: ServiceBlockingGateProps) => {
console.info('[ServiceBlockingGate] Refreshing service + agent status');
try {
const [service, agent] = await Promise.all([
const [serviceResult, agentResult] = await Promise.allSettled([
openhumanServiceStatus(),
openhumanAgentServerStatus(),
]);
const serviceState = service?.result?.state;
const normalized = normalizeServiceState(serviceState);
const normalized =
serviceResult.status === 'fulfilled'
? normalizeServiceState(serviceResult.value?.result?.state)
: 'Unknown';
const serviceRunning = normalized === 'Running';
const agentIsRunning = !!agent?.result?.running;
const agentIsRunning =
agentResult.status === 'fulfilled' ? !!agentResult.value?.result?.running : false;
const gateReady = serviceRunning || agentIsRunning;
if (serviceResult.status !== 'fulfilled' && !agentIsRunning) {
throw serviceResult.reason;
}
setServiceStateText(prev => (prev === normalized ? prev : normalized));
setAgentRunning(prev => (prev === agentIsRunning ? prev : agentIsRunning));
setGateStatus(prev => {
const next = serviceRunning && agentIsRunning ? 'ready' : 'blocked';
const next = gateReady ? 'ready' : 'blocked';
return prev === next ? prev : next;
});
setError(prev => (prev ? null : prev));
console.info('[ServiceBlockingGate] Status refreshed', {
serviceState: normalized,
agentRunning: agentIsRunning,
nextGateStatus: serviceRunning && agentIsRunning ? 'ready' : 'blocked',
nextGateStatus: gateReady ? 'ready' : 'blocked',
passMode: serviceRunning ? 'hard(service)' : agentIsRunning ? 'soft(agent)' : 'blocked',
});
} catch (err) {
setServiceStateText('Unknown');
@@ -183,7 +195,7 @@ const ServiceBlockingGate = ({ children }: ServiceBlockingGateProps) => {
</button>
<button
disabled={isOperating || !installed || (serviceRunning && agentRunning)}
disabled={isOperating || !installed || serviceRunning}
onClick={() => void runOperation(() => openhumanServiceStart())}
className="px-3 py-2 rounded-lg text-sm bg-green-600 hover:bg-green-700 disabled:bg-gray-700 disabled:text-gray-400">
Start Service
app/src/components/daemon/__tests__/ServiceBlockingGate.test.tsx
+60
View File
@@ -57,4 +57,64 @@ describe('ServiceBlockingGate', () => {
fireEvent.click(screen.getByRole('button', { name: 'Install Service' }));
await waitFor(() => expect(mockInstall).toHaveBeenCalled());
});
it('renders children when service is running even if agent is not running', async () => {
mockIsTauri.mockReturnValue(true);
mockServiceStatus.mockResolvedValue({ result: { state: 'Running' }, logs: [] });
mockAgentStatus.mockResolvedValue({ result: { running: false }, logs: [] });
render(
<ServiceBlockingGate>
<div>App Content</div>
</ServiceBlockingGate>
);
await waitFor(() => expect(screen.getByText('App Content')).toBeInTheDocument());
expect(screen.queryByText('OpenHuman Service Required')).not.toBeInTheDocument();
});
it('renders children when service is running and agent probe fails', async () => {
mockIsTauri.mockReturnValue(true);
mockServiceStatus.mockResolvedValue({ result: { state: 'Running' }, logs: [] });
mockAgentStatus.mockRejectedValue(new Error('agent status unavailable'));
render(
<ServiceBlockingGate>
<div>App Content</div>
</ServiceBlockingGate>
);
await waitFor(() => expect(screen.getByText('App Content')).toBeInTheDocument());
expect(screen.queryByText('OpenHuman Service Required')).not.toBeInTheDocument();
});
it('renders children when service is stopped but agent server is running (soft pass)', async () => {
mockIsTauri.mockReturnValue(true);
mockServiceStatus.mockResolvedValue({ result: { state: 'Stopped' }, logs: [] });
mockAgentStatus.mockResolvedValue({ result: { running: true }, logs: [] });
render(
<ServiceBlockingGate>
<div>App Content</div>
</ServiceBlockingGate>
);
await waitFor(() => expect(screen.getByText('App Content')).toBeInTheDocument());
expect(screen.queryByText('OpenHuman Service Required')).not.toBeInTheDocument();
});
it('renders children when service probe fails but agent server is running (soft pass)', async () => {
mockIsTauri.mockReturnValue(true);
mockServiceStatus.mockRejectedValue(new Error('service status unavailable'));
mockAgentStatus.mockResolvedValue({ result: { running: true }, logs: [] });
render(
<ServiceBlockingGate>
<div>App Content</div>
</ServiceBlockingGate>
);
await waitFor(() => expect(screen.getByText('App Content')).toBeInTheDocument());
expect(screen.queryByText('OpenHuman Service Required')).not.toBeInTheDocument();
});
});
app/src/main.tsx
+4
View File
@@ -7,8 +7,12 @@ import ErrorReportNotification from './components/ErrorReportNotification';
import './index.css';
import './polyfills';
import { initSentry } from './services/analytics';
import { setStoreForApiClient } from './services/apiClient';
import { store } from './store';
import { setupDesktopDeepLinkListener } from './utils/desktopDeepLinkListener';
setStoreForApiClient(() => store.getState().auth.token);
const ensureDefaultHashRoute = () => {
const hash = window.location.hash;
if (!hash || hash === '#') {
app/src/services/apiClient.ts
+9 -9
View File
@@ -12,16 +12,16 @@ interface RequestOptions {
}
/**
* Lazy store accessor to break the circular dependency:
* store/index → slices → api services → apiClient → store/index
* Lazy auth token accessor so `apiClient` never imports `store/index` at module level.
* Entry (`main.tsx`) and Vitest setup call `setStoreForApiClient` after the store module loads,
* avoiding a cycle: `store` → `apiClient` → … → `socketService` → `store`.
*
* The store registers itself via `setStoreForApiClient` after creation,
* so apiClient never imports store/index at module level.
* The binding name avoids clashing with transpiled private method names (e.g. `_getToken`).
*/
let _getToken: (() => string | null) | null = null;
let authTokenGetterRef: (() => string | null) | null = null;
export function setStoreForApiClient(getToken: () => string | null) {
_getToken = getToken;
authTokenGetterRef = getToken;
}
/**
@@ -29,8 +29,8 @@ export function setStoreForApiClient(getToken: () => string | null) {
* Handles authentication, error handling, and response typing
*/
class ApiClient {
private getToken(): string | null {
return _getToken ? _getToken() : null;
private resolveAuthToken(): string | null {
return authTokenGetterRef ? authTokenGetterRef() : null;
}
/**
@@ -44,7 +44,7 @@ class ApiClient {
// Add authorization header if auth is required
if (options.requireAuth !== false) {
const token = this.getToken();
const token = this.resolveAuthToken();
if (token) {
headers.Authorization = `Bearer ${token}`;
}
app/src/store/index.ts
-4
View File
@@ -12,7 +12,6 @@ import {
} from 'redux-persist';
import storage from 'redux-persist/lib/storage';
import { setStoreForApiClient } from '../services/apiClient';
import { IS_DEV } from '../utils/config';
import {
logout as clearRustSession,
@@ -152,9 +151,6 @@ export const store = configureStore({
},
});
// Wire up the apiClient so it can read the token without a circular import
setStoreForApiClient(() => store.getState().auth.token);
export const persistor = persistStore(store, null, () => {
// Dev-only: auto-inject JWT token for local testing without login flow.
const devToken = import.meta.env.VITE_DEV_JWT_TOKEN;
src/core/all.rs
+5 -2
View File
@@ -46,7 +46,9 @@ fn build_registered_controllers() -> Vec<RegisteredController> {
controllers.extend(crate::openhuman::heartbeat::all_heartbeat_registered_controllers());
controllers.extend(crate::openhuman::cost::all_cost_registered_controllers());
controllers.extend(crate::openhuman::autocomplete::all_autocomplete_registered_controllers());
controllers.extend(crate::openhuman::web_channel::all_web_channel_registered_controllers());
controllers.extend(
crate::openhuman::channels::providers::web::all_web_channel_registered_controllers(),
);
controllers.extend(crate::openhuman::config::all_config_registered_controllers());
controllers.extend(crate::openhuman::credentials::all_credentials_registered_controllers());
controllers.extend(crate::openhuman::service::all_service_registered_controllers());
@@ -72,7 +74,8 @@ fn build_declared_controller_schemas() -> Vec<ControllerSchema> {
schemas.extend(crate::openhuman::heartbeat::all_heartbeat_controller_schemas());
schemas.extend(crate::openhuman::cost::all_cost_controller_schemas());
schemas.extend(crate::openhuman::autocomplete::all_autocomplete_controller_schemas());
schemas.extend(crate::openhuman::web_channel::all_web_channel_controller_schemas());
schemas
.extend(crate::openhuman::channels::providers::web::all_web_channel_controller_schemas());
schemas.extend(crate::openhuman::config::all_config_controller_schemas());
schemas.extend(crate::openhuman::credentials::all_credentials_controller_schemas());
schemas.extend(crate::openhuman::service::all_service_controller_schemas());
src/core/jsonrpc.rs
+1 -1
View File
@@ -156,7 +156,7 @@ async fn events_handler(
Query(query): Query<EventsQuery>,
) -> Sse<impl tokio_stream::Stream<Item = Result<Event, std::convert::Infallible>>> {
let client_id = query.client_id;
let rx = crate::openhuman::web_channel::subscribe_web_channel_events();
let rx = crate::openhuman::channels::providers::web::subscribe_web_channel_events();
let stream = tokio_stream::wrappers::BroadcastStream::new(rx).filter_map(move |item| {
let event = match item {
Ok(ev) => ev,
src/core/socketio.rs
+84 -12
View File
@@ -1,9 +1,35 @@
use serde::Deserialize;
use serde::Serialize;
use serde_json::json;
use socketioxide::extract::{Data, SocketRef};
use socketioxide::SocketIo;
use crate::openhuman::web_channel::events::WebChannelEvent;
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct WebChannelEvent {
pub event: String,
pub client_id: String,
pub thread_id: String,
pub request_id: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub full_response: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub message: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub error_type: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub tool_name: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub skill_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub args: Option<serde_json::Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub output: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub success: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub round: Option<u32>,
}
#[derive(Debug, Deserialize)]
struct SocketRpcRequest {
@@ -107,7 +133,7 @@ pub fn attach_socketio() -> (socketioxide::layer::SocketIoLayer, SocketIo) {
payload.temperature
);
if let Err(error) = crate::openhuman::web_channel::ops::channel_web_chat(
match crate::openhuman::channels::providers::web::start_chat(
&client_id,
&payload.thread_id,
&payload.message,
@@ -116,7 +142,18 @@ pub fn attach_socketio() -> (socketioxide::layer::SocketIoLayer, SocketIo) {
)
.await
{
let error_payload = json!({
Ok(request_id) => {
let accepted_payload = json!({
"event": "chat_accepted",
"client_id": client_id,
"thread_id": thread_id,
"request_id": request_id,
});
log::debug!("[socketio] send event=chat_accepted client_id={} thread_id={}", socket.id, payload.thread_id);
emit_with_aliases(&socket, "chat_accepted", &accepted_payload);
}
Err(error) => {
let error_payload = json!({
"event": "chat_error",
"client_id": client_id,
"thread_id": thread_id,
@@ -124,8 +161,9 @@ pub fn attach_socketio() -> (socketioxide::layer::SocketIoLayer, SocketIo) {
"message": error,
"error_type": "inference",
});
log::debug!("[socketio] send event=chat_error client_id={} thread_id={} message={}", socket.id, payload.thread_id, error);
let _ = socket.emit("chat_error", &error_payload);
log::debug!("[socketio] send event=chat_error client_id={} thread_id={} message={}", socket.id, payload.thread_id, error);
emit_with_aliases(&socket, "chat_error", &error_payload);
}
}
});
@@ -136,11 +174,9 @@ pub fn attach_socketio() -> (socketioxide::layer::SocketIoLayer, SocketIo) {
client_id,
payload.thread_id
);
let _ = crate::openhuman::web_channel::ops::channel_web_cancel(
&client_id,
&payload.thread_id,
)
.await;
let _ =
crate::openhuman::channels::providers::web::cancel_chat(&client_id, &payload.thread_id)
.await;
});
});
@@ -149,7 +185,7 @@ pub fn attach_socketio() -> (socketioxide::layer::SocketIoLayer, SocketIo) {
pub fn spawn_web_channel_bridge(io: SocketIo) {
tokio::spawn(async move {
let mut rx = crate::openhuman::web_channel::subscribe_web_channel_events();
let mut rx = crate::openhuman::channels::providers::web::subscribe_web_channel_events();
loop {
let event = match rx.recv().await {
Ok(event) => event,
@@ -186,7 +222,31 @@ fn emit_web_channel_event(io: &SocketIo, event: WebChannelEvent) {
.and_then(|v| v.as_str())
.unwrap_or_default()
);
let _ = io.to(room).emit(name, &payload);
emit_room_with_aliases(io, &room, &name, &payload);
}
}
fn event_alias(name: &str) -> Option<String> {
if name.contains('_') {
return Some(name.replace('_', ":"));
}
if name.contains(':') {
return Some(name.replace(':', "_"));
}
None
}
fn emit_with_aliases(socket: &SocketRef, name: &str, payload: &serde_json::Value) {
let _ = socket.emit(name, payload);
if let Some(alias) = event_alias(name) {
let _ = socket.emit(alias, payload);
}
}
fn emit_room_with_aliases(io: &SocketIo, room: &str, name: &str, payload: &serde_json::Value) {
let _ = io.to(room.to_string()).emit(name.to_string(), payload);
if let Some(alias) = event_alias(name) {
let _ = io.to(room.to_string()).emit(alias, payload);
}
}
@@ -200,3 +260,15 @@ fn json_type_name(value: &serde_json::Value) -> &'static str {
serde_json::Value::Object(_) => "object",
}
}
#[cfg(test)]
mod tests {
use super::event_alias;
#[test]
fn event_alias_translates_between_delimiters() {
assert_eq!(event_alias("chat_done").as_deref(), Some("chat:done"));
assert_eq!(event_alias("chat:error").as_deref(), Some("chat_error"));
assert_eq!(event_alias("ready"), None);
}
}
src/openhuman/agent/agent.rs
+5 -47
View File
@@ -396,33 +396,6 @@ impl Agent {
self.model_name.clone()
}
fn has_tool_named(&self, name: &str) -> bool {
self.tools.iter().any(|tool| tool.name() == name)
}
fn should_force_list_files_fallback(&self, user_message: &str, assistant_text: &str) -> bool {
if !self.has_tool_named("shell") {
return false;
}
let user = user_message.to_ascii_lowercase();
let asks_for_files = (user.contains("files") || user.contains("list"))
&& (user.contains("current dir")
|| user.contains("current directory")
|| user.contains("this dir")
|| user.contains("this directory")
|| user.contains("here"));
if !asks_for_files {
return false;
}
let assistant = assistant_text.to_ascii_lowercase();
assistant.contains("let's check")
|| assistant.contains("let me check")
|| assistant.contains("checking")
|| assistant.contains("i'll check")
}
pub async fn turn(&mut self, user_message: &str) -> Result<String> {
log::info!(
"[agent_loop] turn start message_chars={} history_len={} max_tool_iterations={}",
@@ -432,6 +405,11 @@ impl Agent {
);
if self.history.is_empty() {
let system_prompt = self.build_system_prompt()?;
log::info!(
"[agent_loop] system prompt built chars={} content=\n{}",
system_prompt.chars().count(),
system_prompt
);
self.history
.push(ConversationMessage::Chat(ChatMessage::system(
system_prompt,
@@ -514,26 +492,6 @@ impl Agent {
calls.len()
);
if calls.is_empty() {
if iteration == 0 && self.should_force_list_files_fallback(user_message, &text) {
log::warn!(
"[agent_loop] applying list-files direct fallback i={} tool=shell command=ls -la",
iteration + 1
);
let fallback_call = ParsedToolCall {
name: "shell".to_string(),
arguments: serde_json::json!({"command": "ls -la"}),
tool_call_id: None,
};
let result = self.execute_tool_call(&fallback_call).await;
let final_text = result.output;
self.history
.push(ConversationMessage::Chat(ChatMessage::assistant(
final_text.clone(),
)));
self.trim_history();
return Ok(final_text);
}
let final_text = if text.is_empty() {
response.text.unwrap_or_default()
} else {
src/openhuman/agent/schemas.rs
-34
View File
@@ -24,12 +24,6 @@ struct AgentReplSessionStartParams {
temperature: Option<f64>,
}
#[derive(Debug, Deserialize)]
struct AgentReplSessionChatParams {
session_id: String,
message: String,
}
#[derive(Debug, Deserialize)]
struct AgentReplSessionControlParams {
session_id: String,
@@ -40,7 +34,6 @@ pub fn all_controller_schemas() -> Vec<ControllerSchema> {
schemas("chat"),
schemas("chat_simple"),
schemas("repl_session_start"),
schemas("repl_session_chat"),
schemas("repl_session_reset"),
schemas("repl_session_end"),
schemas("server_status"),
@@ -61,10 +54,6 @@ pub fn all_registered_controllers() -> Vec<RegisteredController> {
schema: schemas("repl_session_start"),
handler: handle_repl_session_start,
},
RegisteredController {
schema: schemas("repl_session_chat"),
handler: handle_repl_session_chat,
},
RegisteredController {
schema: schemas("repl_session_reset"),
handler: handle_repl_session_reset,
@@ -115,16 +104,6 @@ pub fn schemas(function: &str) -> ControllerSchema {
],
outputs: vec![json_output("result", "Session creation result.")],
},
"repl_session_chat" => ControllerSchema {
namespace: "agent",
function: "repl_session_chat",
description: "Send a message through REPL agent session.",
inputs: vec![
required_string("session_id", "REPL session id."),
required_string("message", "User message."),
],
outputs: vec![json_output("response", "Session chat response.")],
},
"repl_session_reset" => ControllerSchema {
namespace: "agent",
function: "repl_session_reset",
@@ -209,19 +188,6 @@ fn handle_repl_session_start(params: Map<String, Value>) -> ControllerFuture {
})
}
fn handle_repl_session_chat(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let p = deserialize_params::<AgentReplSessionChatParams>(params)?;
to_json(
crate::openhuman::local_ai::rpc::agent_repl_session_chat(
p.session_id.trim(),
&p.message,
)
.await?,
)
})
}
fn handle_repl_session_reset(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let p = deserialize_params::<AgentReplSessionControlParams>(params)?;
src/openhuman/channels/mod.rs
+1
View File
@@ -28,6 +28,7 @@ pub use providers::qq;
pub use providers::signal;
pub use providers::slack;
pub use providers::telegram;
pub use providers::web;
pub use providers::whatsapp;
#[cfg(feature = "whatsapp-web")]
pub use providers::whatsapp_storage;
src/openhuman/channels/providers/mod.rs
+1
View File
@@ -14,6 +14,7 @@ pub mod qq;
pub mod signal;
pub mod slack;
pub mod telegram;
pub mod web;
pub mod whatsapp;
#[cfg(feature = "whatsapp-web")]
pub mod whatsapp_storage;
src/openhuman/channels/providers/web.rs
+604
View File
@@ -0,0 +1,604 @@
use once_cell::sync::Lazy;
use serde::Deserialize;
use serde_json::{json, Map, Value};
use std::collections::HashMap;
use tokio::sync::{broadcast, Mutex};
use uuid::Uuid;
use crate::core::all::{ControllerFuture, RegisteredController};
use crate::core::socketio::WebChannelEvent;
use crate::core::{ControllerSchema, FieldSchema, TypeSchema};
use crate::openhuman::agent::Agent;
use crate::openhuman::config::rpc as config_rpc;
use crate::openhuman::config::Config;
use crate::openhuman::providers::ConversationMessage;
use crate::rpc::RpcOutcome;
static EVENT_BUS: Lazy<broadcast::Sender<WebChannelEvent>> = Lazy::new(|| {
let (tx, _rx) = broadcast::channel(512);
tx
});
pub fn subscribe_web_channel_events() -> broadcast::Receiver<WebChannelEvent> {
EVENT_BUS.subscribe()
}
pub fn publish_web_channel_event(event: WebChannelEvent) {
let _ = EVENT_BUS.send(event);
}
struct SessionEntry {
agent: Agent,
model_override: Option<String>,
temperature: Option<f64>,
}
#[derive(Debug)]
struct InFlightEntry {
request_id: String,
handle: tokio::task::JoinHandle<()>,
}
static THREAD_SESSIONS: Lazy<Mutex<HashMap<String, SessionEntry>>> =
Lazy::new(|| Mutex::new(HashMap::new()));
static IN_FLIGHT: Lazy<Mutex<HashMap<String, InFlightEntry>>> =
Lazy::new(|| Mutex::new(HashMap::new()));
fn key_for(client_id: &str, thread_id: &str) -> String {
format!("{client_id}::{thread_id}")
}
pub async fn start_chat(
client_id: &str,
thread_id: &str,
message: &str,
model_override: Option<String>,
temperature: Option<f64>,
) -> Result<String, String> {
let client_id = client_id.trim().to_string();
let thread_id = thread_id.trim().to_string();
let message = message.trim().to_string();
if client_id.is_empty() {
return Err("client_id is required".to_string());
}
if thread_id.is_empty() {
return Err("thread_id is required".to_string());
}
if message.is_empty() {
return Err("message is required".to_string());
}
let request_id = Uuid::new_v4().to_string();
let map_key = key_for(&client_id, &thread_id);
{
let mut in_flight = IN_FLIGHT.lock().await;
if let Some(existing) = in_flight.remove(&map_key) {
existing.handle.abort();
publish_web_channel_event(WebChannelEvent {
event: "chat_error".to_string(),
client_id: client_id.clone(),
thread_id: thread_id.clone(),
request_id: existing.request_id,
full_response: None,
message: Some("Cancelled by newer request".to_string()),
error_type: Some("cancelled".to_string()),
tool_name: None,
skill_id: None,
args: None,
output: None,
success: None,
round: None,
});
}
}
let client_id_task = client_id.clone();
let thread_id_task = thread_id.clone();
let request_id_task = request_id.clone();
let map_key_task = map_key.clone();
let handle = tokio::spawn(async move {
let result = run_chat_task(
&client_id_task,
&thread_id_task,
&request_id_task,
&message,
model_override,
temperature,
)
.await;
match result {
Ok(full_response) => {
publish_web_channel_event(WebChannelEvent {
event: "chat_done".to_string(),
client_id: client_id_task.clone(),
thread_id: thread_id_task.clone(),
request_id: request_id_task.clone(),
full_response: Some(full_response),
message: None,
error_type: None,
tool_name: None,
skill_id: None,
args: None,
output: None,
success: None,
round: None,
});
}
Err(err) => {
publish_web_channel_event(WebChannelEvent {
event: "chat_error".to_string(),
client_id: client_id_task.clone(),
thread_id: thread_id_task.clone(),
request_id: request_id_task.clone(),
full_response: None,
message: Some(err),
error_type: Some("inference".to_string()),
tool_name: None,
skill_id: None,
args: None,
output: None,
success: None,
round: None,
});
}
}
let mut in_flight = IN_FLIGHT.lock().await;
if let Some(current) = in_flight.get(&map_key_task) {
if current.request_id == request_id_task {
in_flight.remove(&map_key_task);
}
}
});
{
let mut in_flight = IN_FLIGHT.lock().await;
in_flight.insert(
map_key,
InFlightEntry {
request_id: request_id.clone(),
handle,
},
);
}
Ok(request_id)
}
pub async fn cancel_chat(client_id: &str, thread_id: &str) -> Result<Option<String>, String> {
let client_id = client_id.trim();
let thread_id = thread_id.trim();
if client_id.is_empty() {
return Err("client_id is required".to_string());
}
if thread_id.is_empty() {
return Err("thread_id is required".to_string());
}
let map_key = key_for(client_id, thread_id);
let mut removed_request_id: Option<String> = None;
{
let mut in_flight = IN_FLIGHT.lock().await;
if let Some(existing) = in_flight.remove(&map_key) {
removed_request_id = Some(existing.request_id.clone());
existing.handle.abort();
}
}
if let Some(request_id) = removed_request_id.clone() {
publish_web_channel_event(WebChannelEvent {
event: "chat_error".to_string(),
client_id: client_id.to_string(),
thread_id: thread_id.to_string(),
request_id,
full_response: None,
message: Some("Cancelled".to_string()),
error_type: Some("cancelled".to_string()),
tool_name: None,
skill_id: None,
args: None,
output: None,
success: None,
round: None,
});
}
Ok(removed_request_id)
}
async fn run_chat_task(
client_id: &str,
thread_id: &str,
request_id: &str,
message: &str,
model_override: Option<String>,
temperature: Option<f64>,
) -> Result<String, String> {
let config = config_rpc::load_config_with_timeout().await?;
let map_key = key_for(client_id, thread_id);
let model_override = normalize_model_override(model_override);
let prior = {
let mut sessions = THREAD_SESSIONS.lock().await;
sessions.remove(&map_key)
};
let mut agent = match prior {
Some(entry)
if entry.model_override == model_override && entry.temperature == temperature =>
{
entry.agent
}
Some(_) | None => build_session_agent(&config, model_override.clone(), temperature)?,
};
let history_before = agent.history().len();
let result = agent.run_single(message).await.map_err(|e| e.to_string());
if result.is_ok() {
publish_tool_events_from_history(
client_id,
thread_id,
request_id,
&agent.history()[history_before..],
);
}
{
let mut sessions = THREAD_SESSIONS.lock().await;
sessions.insert(
map_key,
SessionEntry {
agent,
model_override,
temperature,
},
);
}
result
}
fn publish_tool_events_from_history(
client_id: &str,
thread_id: &str,
request_id: &str,
messages: &[ConversationMessage],
) {
let mut round: u32 = 0;
let mut current_round_calls: Vec<(String, String)> = Vec::new();
for message in messages {
match message {
ConversationMessage::AssistantToolCalls { tool_calls, .. } => {
round += 1;
current_round_calls.clear();
for (idx, call) in tool_calls.iter().enumerate() {
let synthetic_id = if call.id.trim().is_empty() {
format!("idx:{idx}")
} else {
call.id.clone()
};
current_round_calls.push((synthetic_id, call.name.clone()));
publish_web_channel_event(WebChannelEvent {
event: "tool_call".to_string(),
client_id: client_id.to_string(),
thread_id: thread_id.to_string(),
request_id: request_id.to_string(),
full_response: None,
message: None,
error_type: None,
tool_name: Some(call.name.clone()),
skill_id: Some("web_channel".to_string()),
args: Some(parse_tool_args(&call.arguments)),
output: None,
success: None,
round: Some(round),
});
}
}
ConversationMessage::ToolResults(results) => {
for (idx, result) in results.iter().enumerate() {
let fallback = format!("idx:{idx}");
let tool_name = current_round_calls
.iter()
.find(|(tool_call_id, _)| tool_call_id == &result.tool_call_id)
.or_else(|| current_round_calls.get(idx))
.map(|(_, name)| name.clone())
.unwrap_or_else(|| "unknown".to_string());
let success = !result.content.trim_start().starts_with("Error:");
publish_web_channel_event(WebChannelEvent {
event: "tool_result".to_string(),
client_id: client_id.to_string(),
thread_id: thread_id.to_string(),
request_id: request_id.to_string(),
full_response: None,
message: None,
error_type: None,
tool_name: Some(tool_name),
skill_id: Some("web_channel".to_string()),
args: Some(Value::Object(Map::from_iter([(
"tool_call_id".to_string(),
Value::String(if result.tool_call_id.is_empty() {
fallback
} else {
result.tool_call_id.clone()
}),
)]))),
output: Some(result.content.clone()),
success: Some(success),
round: Some(round.max(1)),
});
}
}
ConversationMessage::Chat(_) => {}
}
}
}
fn parse_tool_args(arguments: &str) -> Value {
if arguments.trim().is_empty() {
return Value::Object(Map::new());
}
match serde_json::from_str::<Value>(arguments) {
Ok(value) => value,
Err(_) => Value::Object(Map::from_iter([(
"raw".to_string(),
Value::String(arguments.to_string()),
)])),
}
}
fn normalize_model_override(model_override: Option<String>) -> Option<String> {
model_override
.map(|model| model.trim().to_string())
.filter(|model| !model.is_empty())
}
fn build_session_agent(
config: &Config,
model_override: Option<String>,
temperature: Option<f64>,
) -> Result<Agent, String> {
let mut effective = config.clone();
if let Some(model) = model_override {
effective.default_model = Some(model);
}
if let Some(temp) = temperature {
effective.default_temperature = temp;
}
Agent::from_config(&effective).map_err(|e| e.to_string())
}
#[derive(Debug, Deserialize)]
struct WebChatParams {
client_id: String,
thread_id: String,
message: String,
model_override: Option<String>,
temperature: Option<f64>,
}
#[derive(Debug, Deserialize)]
struct WebCancelParams {
client_id: String,
thread_id: String,
}
pub async fn channel_web_chat(
client_id: &str,
thread_id: &str,
message: &str,
model_override: Option<String>,
temperature: Option<f64>,
) -> Result<RpcOutcome<Value>, String> {
let request_id = start_chat(client_id, thread_id, message, model_override, temperature).await?;
Ok(RpcOutcome::single_log(
json!({
"accepted": true,
"client_id": client_id.trim(),
"thread_id": thread_id.trim(),
"request_id": request_id,
}),
"web channel request accepted",
))
}
pub async fn channel_web_cancel(
client_id: &str,
thread_id: &str,
) -> Result<RpcOutcome<Value>, String> {
let cancelled_request_id = cancel_chat(client_id, thread_id).await?;
Ok(RpcOutcome::single_log(
json!({
"cancelled": cancelled_request_id.is_some(),
"client_id": client_id.trim(),
"thread_id": thread_id.trim(),
"request_id": cancelled_request_id,
}),
"web channel cancellation processed",
))
}
pub fn all_web_channel_controller_schemas() -> Vec<ControllerSchema> {
vec![schemas("chat"), schemas("cancel")]
}
pub fn all_web_channel_registered_controllers() -> Vec<RegisteredController> {
vec![
RegisteredController {
schema: schemas("chat"),
handler: handle_chat,
},
RegisteredController {
schema: schemas("cancel"),
handler: handle_cancel,
},
]
}
pub fn schemas(function: &str) -> ControllerSchema {
match function {
"chat" => ControllerSchema {
namespace: "channel",
function: "web_chat",
description: "Send a web channel message through the agent loop.",
inputs: vec![
required_string("client_id", "Client stream identifier."),
required_string("thread_id", "Thread identifier."),
required_string("message", "User message."),
optional_string("model_override", "Optional model override."),
optional_f64("temperature", "Optional temperature override."),
],
outputs: vec![json_output("ack", "Acceptance payload.")],
},
"cancel" => ControllerSchema {
namespace: "channel",
function: "web_cancel",
description: "Cancel in-flight web channel request for a thread.",
inputs: vec![
required_string("client_id", "Client stream identifier."),
required_string("thread_id", "Thread identifier."),
],
outputs: vec![json_output("ack", "Cancellation payload.")],
},
_ => ControllerSchema {
namespace: "channel",
function: "unknown",
description: "Unknown web channel controller function.",
inputs: vec![],
outputs: vec![FieldSchema {
name: "error",
ty: TypeSchema::String,
comment: "Lookup error details.",
required: true,
}],
},
}
}
fn handle_chat(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let p = deserialize_params::<WebChatParams>(params)?;
to_json(
channel_web_chat(
&p.client_id,
&p.thread_id,
&p.message,
p.model_override,
p.temperature,
)
.await?,
)
})
}
fn handle_cancel(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let p = deserialize_params::<WebCancelParams>(params)?;
to_json(channel_web_cancel(&p.client_id, &p.thread_id).await?)
})
}
fn deserialize_params<T: serde::de::DeserializeOwned>(
params: Map<String, Value>,
) -> Result<T, String> {
serde_json::from_value(Value::Object(params)).map_err(|e| format!("invalid params: {e}"))
}
fn required_string(name: &'static str, comment: &'static str) -> FieldSchema {
FieldSchema {
name,
ty: TypeSchema::String,
comment,
required: true,
}
}
fn optional_string(name: &'static str, comment: &'static str) -> FieldSchema {
FieldSchema {
name,
ty: TypeSchema::Option(Box::new(TypeSchema::String)),
comment,
required: false,
}
}
fn optional_f64(name: &'static str, comment: &'static str) -> FieldSchema {
FieldSchema {
name,
ty: TypeSchema::Option(Box::new(TypeSchema::F64)),
comment,
required: false,
}
}
fn json_output(name: &'static str, comment: &'static str) -> FieldSchema {
FieldSchema {
name,
ty: TypeSchema::Json,
comment,
required: true,
}
}
fn to_json<T: serde::Serialize>(outcome: RpcOutcome<T>) -> Result<Value, String> {
outcome.into_cli_compatible_json()
}
#[cfg(test)]
mod tests {
use super::{cancel_chat, parse_tool_args, start_chat};
use serde_json::json;
#[tokio::test]
async fn start_chat_validates_required_fields() {
let err = start_chat("", "thread", "hello", None, None)
.await
.expect_err("client id should be required");
assert!(err.contains("client_id is required"));
let err = start_chat("client", "", "hello", None, None)
.await
.expect_err("thread id should be required");
assert!(err.contains("thread_id is required"));
let err = start_chat("client", "thread", " ", None, None)
.await
.expect_err("message should be required");
assert!(err.contains("message is required"));
}
#[tokio::test]
async fn cancel_chat_validates_required_fields() {
let err = cancel_chat("", "thread")
.await
.expect_err("client id should be required");
assert!(err.contains("client_id is required"));
let err = cancel_chat("client", "")
.await
.expect_err("thread id should be required");
assert!(err.contains("thread_id is required"));
}
#[test]
fn parse_tool_args_handles_json_and_raw_fallback() {
assert_eq!(
parse_tool_args(r#"{"command":"date"}"#),
json!({"command":"date"})
);
assert_eq!(parse_tool_args(""), json!({}));
assert_eq!(parse_tool_args("not-json"), json!({"raw":"not-json"}));
}
}
src/openhuman/local_ai/ops.rs
-54
View File
@@ -122,60 +122,6 @@ pub async fn agent_repl_session_start(
))
}
pub async fn agent_repl_session_chat(
session_id: &str,
message: &str,
) -> Result<RpcOutcome<String>, String> {
let session_id = session_id.trim();
if session_id.is_empty() {
return Err("session_id is required".to_string());
}
let started = std::time::Instant::now();
log::info!(
"[agent_repl] chat start session_id={} message_chars={}",
session_id,
message.chars().count()
);
let mut agent = {
let mut sessions = REPL_AGENT_SESSIONS.lock().await;
sessions
.remove(session_id)
.ok_or_else(|| format!("agent repl session not found: {session_id}"))?
};
let result = agent.run_single(message).await;
REPL_AGENT_SESSIONS
.lock()
.await
.insert(session_id.to_string(), agent);
let response = match result {
Ok(response) => {
log::info!(
"[agent_repl] chat ok session_id={} elapsed_ms={} response_chars={}",
session_id,
started.elapsed().as_millis(),
response.chars().count()
);
response
}
Err(err) => {
log::error!(
"[agent_repl] chat error session_id={} elapsed_ms={} error={}",
session_id,
started.elapsed().as_millis(),
err
);
return Err(err.to_string());
}
};
Ok(RpcOutcome::single_log(
response,
"agent repl session chat completed",
))
}
pub async fn agent_repl_session_reset(
session_id: &str,
) -> Result<RpcOutcome<serde_json::Value>, String> {
src/openhuman/local_ai/schemas.rs
-34
View File
@@ -24,12 +24,6 @@ struct AgentReplSessionStartParams {
temperature: Option<f64>,
}
#[derive(Debug, Deserialize)]
struct AgentReplSessionChatParams {
session_id: String,
message: String,
}
#[derive(Debug, Deserialize)]
struct AgentReplSessionControlParams {
session_id: String,
@@ -100,7 +94,6 @@ pub fn all_controller_schemas() -> Vec<ControllerSchema> {
schemas("agent_chat"),
schemas("agent_chat_simple"),
schemas("agent_repl_session_start"),
schemas("agent_repl_session_chat"),
schemas("agent_repl_session_reset"),
schemas("agent_repl_session_end"),
schemas("local_ai_status"),
@@ -134,10 +127,6 @@ pub fn all_registered_controllers() -> Vec<RegisteredController> {
schema: schemas("agent_repl_session_start"),
handler: handle_agent_repl_session_start,
},
RegisteredController {
schema: schemas("agent_repl_session_chat"),
handler: handle_agent_repl_session_chat,
},
RegisteredController {
schema: schemas("agent_repl_session_reset"),
handler: handle_agent_repl_session_reset,
@@ -240,16 +229,6 @@ pub fn schemas(function: &str) -> ControllerSchema {
],
outputs: vec![json_output("result", "Session creation result.")],
},
"agent_repl_session_chat" => ControllerSchema {
namespace: "local_ai",
function: "agent_repl_session_chat",
description: "Send a message through REPL agent session.",
inputs: vec![
required_string("session_id", "REPL session id."),
required_string("message", "User message."),
],
outputs: vec![json_output("response", "Session chat response.")],
},
"agent_repl_session_reset" => ControllerSchema {
namespace: "local_ai",
function: "agent_repl_session_reset",
@@ -467,19 +446,6 @@ fn handle_agent_repl_session_start(params: Map<String, Value>) -> ControllerFutu
})
}
fn handle_agent_repl_session_chat(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let p = deserialize_params::<AgentReplSessionChatParams>(params)?;
to_json(
crate::openhuman::local_ai::rpc::agent_repl_session_chat(
p.session_id.trim(),
&p.message,
)
.await?,
)
})
}
fn handle_agent_repl_session_reset(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let p = deserialize_params::<AgentReplSessionControlParams>(params)?;
src/openhuman/mod.rs
-1
View File
@@ -36,5 +36,4 @@ pub mod tools;
pub mod tray;
pub mod tunnel;
pub mod util;
pub mod web_channel;
pub mod workspace;
src/openhuman/service/macos.rs
+37 -24
View File
@@ -197,12 +197,7 @@ pub(crate) fn stop(_config: &Config) -> Result<()> {
}
pub(crate) fn status(_config: &Config) -> Result<ServiceStatus> {
let out = run_capture(Command::new("launchctl").arg("list"))?;
let running = out.lines().any(|line| {
line.contains(SERVICE_LABEL)
|| line.contains(LEGACY_SERVICE_LABEL)
|| line.contains(LEGACY_APP_LABEL)
});
let running = is_service_running_macos()?;
Ok(ServiceStatus {
state: if running {
ServiceState::Running
@@ -263,26 +258,44 @@ fn validate_macos_plist(path: &std::path::Path) -> Result<()> {
}
fn is_service_loaded_macos() -> Result<bool> {
if run_check_silent(
Command::new("launchctl")
.arg("print")
.arg(macos_target(SERVICE_LABEL)?),
) {
return Ok(true);
for target in candidate_macos_targets(SERVICE_LABEL)? {
if run_check_silent(Command::new("launchctl").arg("print").arg(target)) {
return Ok(true);
}
}
if run_check_silent(
Command::new("launchctl")
.arg("print")
.arg(macos_target(LEGACY_SERVICE_LABEL)?),
) {
return Ok(true);
for target in candidate_macos_targets(LEGACY_SERVICE_LABEL)? {
if run_check_silent(Command::new("launchctl").arg("print").arg(target)) {
return Ok(true);
}
}
if run_check_silent(
Command::new("launchctl")
.arg("print")
.arg(macos_target(LEGACY_APP_LABEL)?),
) {
return Ok(true);
for target in candidate_macos_targets(LEGACY_APP_LABEL)? {
if run_check_silent(Command::new("launchctl").arg("print").arg(target)) {
return Ok(true);
}
}
Ok(false)
}
fn is_service_running_macos() -> Result<bool> {
if is_service_loaded_macos()? {
return Ok(true);
}
// Fallback for environments where `launchctl print` is restricted.
let out = run_capture(Command::new("launchctl").arg("list"))?;
Ok(out.lines().any(|line| {
line.contains(SERVICE_LABEL)
|| line.contains(LEGACY_SERVICE_LABEL)
|| line.contains(LEGACY_APP_LABEL)
}))
}
fn candidate_macos_targets(label: &str) -> Result<Vec<String>> {
let uid = run_capture(Command::new("id").arg("-u"))?;
let uid = uid.trim();
Ok(vec![
format!("gui/{uid}/{label}"),
format!("user/{uid}/{label}"),
format!("system/{label}"),
])
}
src/openhuman/web_channel/events.rs
-31
View File
@@ -1,31 +0,0 @@
use once_cell::sync::Lazy;
use serde::{Deserialize, Serialize};
use tokio::sync::broadcast;
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub struct WebChannelEvent {
pub event: String,
pub client_id: String,
pub thread_id: String,
pub request_id: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub full_response: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub message: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub error_type: Option<String>,
}
static EVENT_BUS: Lazy<broadcast::Sender<WebChannelEvent>> = Lazy::new(|| {
let (tx, _rx) = broadcast::channel(512);
tx
});
pub fn subscribe() -> broadcast::Receiver<WebChannelEvent> {
EVENT_BUS.subscribe()
}
pub fn publish(event: WebChannelEvent) {
let _ = EVENT_BUS.send(event);
}
src/openhuman/web_channel/mod.rs
-9
View File
@@ -1,9 +0,0 @@
pub mod events;
pub mod ops;
mod schemas;
pub use events::{publish as publish_web_channel_event, subscribe as subscribe_web_channel_events};
pub use schemas::{
all_controller_schemas as all_web_channel_controller_schemas,
all_registered_controllers as all_web_channel_registered_controllers,
};
src/openhuman/web_channel/ops.rs
-236
View File
@@ -1,236 +0,0 @@
use once_cell::sync::Lazy;
use serde_json::json;
use std::collections::HashMap;
use tokio::sync::Mutex;
use uuid::Uuid;
use crate::openhuman::config::rpc as config_rpc;
use crate::openhuman::local_ai;
use crate::rpc::RpcOutcome;
use super::events::{publish, WebChannelEvent};
#[derive(Debug)]
struct SessionEntry {
session_id: String,
model_override: Option<String>,
temperature: Option<f64>,
}
#[derive(Debug)]
struct InFlightEntry {
request_id: String,
handle: tokio::task::JoinHandle<()>,
}
static THREAD_SESSIONS: Lazy<Mutex<HashMap<String, SessionEntry>>> =
Lazy::new(|| Mutex::new(HashMap::new()));
static IN_FLIGHT: Lazy<Mutex<HashMap<String, InFlightEntry>>> =
Lazy::new(|| Mutex::new(HashMap::new()));
fn key_for(client_id: &str, thread_id: &str) -> String {
format!("{client_id}::{thread_id}")
}
pub async fn channel_web_chat(
client_id: &str,
thread_id: &str,
message: &str,
model_override: Option<String>,
temperature: Option<f64>,
) -> Result<RpcOutcome<serde_json::Value>, String> {
let client_id = client_id.trim().to_string();
let thread_id = thread_id.trim().to_string();
let message = message.trim().to_string();
if client_id.is_empty() {
return Err("client_id is required".to_string());
}
if thread_id.is_empty() {
return Err("thread_id is required".to_string());
}
if message.is_empty() {
return Err("message is required".to_string());
}
let request_id = Uuid::new_v4().to_string();
let map_key = key_for(&client_id, &thread_id);
{
let mut in_flight = IN_FLIGHT.lock().await;
if let Some(existing) = in_flight.remove(&map_key) {
existing.handle.abort();
publish(WebChannelEvent {
event: "chat_error".to_string(),
client_id: client_id.clone(),
thread_id: thread_id.clone(),
request_id: existing.request_id,
full_response: None,
message: Some("Cancelled by newer request".to_string()),
error_type: Some("cancelled".to_string()),
});
}
}
let client_id_task = client_id.clone();
let thread_id_task = thread_id.clone();
let request_id_task = request_id.clone();
let map_key_task = map_key.clone();
let handle = tokio::spawn(async move {
let result = run_chat_task(
&client_id_task,
&thread_id_task,
&message,
model_override,
temperature,
)
.await;
match result {
Ok(full_response) => {
publish(WebChannelEvent {
event: "chat_done".to_string(),
client_id: client_id_task.clone(),
thread_id: thread_id_task.clone(),
request_id: request_id_task.clone(),
full_response: Some(full_response),
message: None,
error_type: None,
});
}
Err(err) => {
publish(WebChannelEvent {
event: "chat_error".to_string(),
client_id: client_id_task.clone(),
thread_id: thread_id_task.clone(),
request_id: request_id_task.clone(),
full_response: None,
message: Some(err),
error_type: Some("inference".to_string()),
});
}
}
let mut in_flight = IN_FLIGHT.lock().await;
if let Some(current) = in_flight.get(&map_key_task) {
if current.request_id == request_id_task {
in_flight.remove(&map_key_task);
}
}
});
{
let mut in_flight = IN_FLIGHT.lock().await;
in_flight.insert(
map_key,
InFlightEntry {
request_id: request_id.clone(),
handle,
},
);
}
Ok(RpcOutcome::single_log(
json!({
"accepted": true,
"client_id": client_id,
"thread_id": thread_id,
"request_id": request_id,
}),
"web channel request accepted",
))
}
pub async fn channel_web_cancel(
client_id: &str,
thread_id: &str,
) -> Result<RpcOutcome<serde_json::Value>, String> {
let client_id = client_id.trim();
let thread_id = thread_id.trim();
if client_id.is_empty() {
return Err("client_id is required".to_string());
}
if thread_id.is_empty() {
return Err("thread_id is required".to_string());
}
let map_key = key_for(client_id, thread_id);
let mut removed_request_id: Option<String> = None;
{
let mut in_flight = IN_FLIGHT.lock().await;
if let Some(existing) = in_flight.remove(&map_key) {
removed_request_id = Some(existing.request_id.clone());
existing.handle.abort();
}
}
if let Some(request_id) = removed_request_id.clone() {
publish(WebChannelEvent {
event: "chat_error".to_string(),
client_id: client_id.to_string(),
thread_id: thread_id.to_string(),
request_id,
full_response: None,
message: Some("Cancelled".to_string()),
error_type: Some("cancelled".to_string()),
});
}
Ok(RpcOutcome::single_log(
json!({
"cancelled": removed_request_id.is_some(),
"client_id": client_id,
"thread_id": thread_id,
}),
"web channel cancellation processed",
))
}
async fn run_chat_task(
client_id: &str,
thread_id: &str,
message: &str,
model_override: Option<String>,
temperature: Option<f64>,
) -> Result<String, String> {
let config = config_rpc::load_config_with_timeout().await?;
let map_key = key_for(client_id, thread_id);
let session_id = format!("web:{client_id}:{thread_id}");
let needs_start = {
let sessions = THREAD_SESSIONS.lock().await;
if let Some(existing) = sessions.get(&map_key) {
existing.model_override != model_override || existing.temperature != temperature
} else {
true
}
};
if needs_start {
let _ = local_ai::rpc::agent_repl_session_end(session_id.as_str()).await;
let _ = local_ai::rpc::agent_repl_session_start(
&config,
Some(session_id.clone()),
model_override.clone(),
temperature,
)
.await?;
let mut sessions = THREAD_SESSIONS.lock().await;
sessions.insert(
map_key,
SessionEntry {
session_id: session_id.clone(),
model_override,
temperature,
},
);
}
let response = local_ai::rpc::agent_repl_session_chat(session_id.as_str(), message).await?;
Ok(response.value)
}
src/openhuman/web_channel/schemas.rs
-148
View File
@@ -1,148 +0,0 @@
use serde::de::DeserializeOwned;
use serde::Deserialize;
use serde_json::{Map, Value};
use crate::core::all::{ControllerFuture, RegisteredController};
use crate::core::{ControllerSchema, FieldSchema, TypeSchema};
#[derive(Debug, Deserialize)]
struct WebChatParams {
client_id: String,
thread_id: String,
message: String,
model_override: Option<String>,
temperature: Option<f64>,
}
#[derive(Debug, Deserialize)]
struct WebCancelParams {
client_id: String,
thread_id: String,
}
pub fn all_controller_schemas() -> Vec<ControllerSchema> {
vec![schemas("chat"), schemas("cancel")]
}
pub fn all_registered_controllers() -> Vec<RegisteredController> {
vec![
RegisteredController {
schema: schemas("chat"),
handler: handle_chat,
},
RegisteredController {
schema: schemas("cancel"),
handler: handle_cancel,
},
]
}
pub fn schemas(function: &str) -> ControllerSchema {
match function {
"chat" => ControllerSchema {
namespace: "channel",
function: "web_chat",
description: "Send a web channel message through the agent loop.",
inputs: vec![
required_string("client_id", "Client stream identifier."),
required_string("thread_id", "Thread identifier."),
required_string("message", "User message."),
optional_string("model_override", "Optional model override."),
optional_f64("temperature", "Optional temperature override."),
],
outputs: vec![json_output("ack", "Acceptance payload.")],
},
"cancel" => ControllerSchema {
namespace: "channel",
function: "web_cancel",
description: "Cancel in-flight web channel request for a thread.",
inputs: vec![
required_string("client_id", "Client stream identifier."),
required_string("thread_id", "Thread identifier."),
],
outputs: vec![json_output("ack", "Cancellation payload.")],
},
_ => ControllerSchema {
namespace: "channel",
function: "unknown",
description: "Unknown web channel controller function.",
inputs: vec![],
outputs: vec![FieldSchema {
name: "error",
ty: TypeSchema::String,
comment: "Lookup error details.",
required: true,
}],
},
}
}
fn handle_chat(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let p = deserialize_params::<WebChatParams>(params)?;
to_json(
crate::openhuman::web_channel::ops::channel_web_chat(
&p.client_id,
&p.thread_id,
&p.message,
p.model_override,
p.temperature,
)
.await?,
)
})
}
fn handle_cancel(params: Map<String, Value>) -> ControllerFuture {
Box::pin(async move {
let p = deserialize_params::<WebCancelParams>(params)?;
to_json(
crate::openhuman::web_channel::ops::channel_web_cancel(&p.client_id, &p.thread_id)
.await?,
)
})
}
fn deserialize_params<T: DeserializeOwned>(params: Map<String, Value>) -> Result<T, String> {
serde_json::from_value(Value::Object(params)).map_err(|e| format!("invalid params: {e}"))
}
fn required_string(name: &'static str, comment: &'static str) -> FieldSchema {
FieldSchema {
name,
ty: TypeSchema::String,
comment,
required: true,
}
}
fn optional_string(name: &'static str, comment: &'static str) -> FieldSchema {
FieldSchema {
name,
ty: TypeSchema::Option(Box::new(TypeSchema::String)),
comment,
required: false,
}
}
fn optional_f64(name: &'static str, comment: &'static str) -> FieldSchema {
FieldSchema {
name,
ty: TypeSchema::Option(Box::new(TypeSchema::F64)),
comment,
required: false,
}
}
fn json_output(name: &'static str, comment: &'static str) -> FieldSchema {
FieldSchema {
name,
ty: TypeSchema::Json,
comment,
required: true,
}
}
fn to_json<T: serde::Serialize>(outcome: crate::rpc::RpcOutcome<T>) -> Result<Value, String> {
outcome.into_cli_compatible_json()
}