gitops/optimclaw
1
0
Fork 0
mirror of https://github.com/outbackdingo/optimclaw.git synced 2026-08-26 23:50:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: /status ownership, model persistence scoping, heartbeat robustness

Browse Source 
Addresses second round of PR review on #1614:

- /status <job_id> DB path now validates job.user_id == requesting user
  before returning data (was missing ownership check, security fix).

- persist_selected_model takes user_id param instead of owner_id, and
  skips .env/TOML writes in multi-tenant mode (these are shared global
  files). handle_system_command now receives user_id from caller.

- JoinSet collection handles Err(JoinError) explicitly instead of
  silently dropping panicked tasks.

- Notification forwarder extracts owner_id from response metadata in
  multi-tenant mode for per-user routing instead of broadcasting to
  the agent owner.

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
This commit is contained in:
[email protected]
2026-03-24 08:58:48 -07:00
co-authored by Claude Opus 4.6
parent 6dfe246288
commit 0d168fb644
3 changed files with 73 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agent/agent_loop.rs
+37 -17
View File
@@ -538,30 +538,50 @@ impl Agent {
.await;
let notify_user = heartbeat_notify_user;
let channels = self.channels.clone();
let is_multi_tenant = hb_config.multi_tenant;
tokio::spawn(async move {
while let Some(response) = notify_rx.recv().await {
// In multi-tenant mode, extract the owning user_id from
// the response metadata so notifications reach the
// correct user rather than the agent's owner.
let effective_user = if is_multi_tenant {
response
.metadata
.get("owner_id")
.and_then(|v| v.as_str())
.map(String::from)
} else {
None
};
// Try the configured channel first, fall back to
// broadcasting on all channels.
let targeted_ok = if let Some(ref channel) = notify_channel
&& let Some(ref user) = notify_target
{
channels
.broadcast(channel, user, response.clone())
.await
.is_ok()
let targeted_ok = if let Some(ref channel) = notify_channel {
let target = effective_user.as_deref().or(notify_target.as_deref());
if let Some(user) = target {
channels
.broadcast(channel, user, response.clone())
.await
.is_ok()
} else {
false
}
} else {
false
};
if !targeted_ok && let Some(ref user) = notify_user {
let results = channels.broadcast_all(user, response).await;
for (ch, result) in results {
if let Err(e) = result {
tracing::warn!(
"Failed to broadcast heartbeat to {}: {}",
ch,
e
);
if !targeted_ok {
let fallback = effective_user.as_deref().or(notify_user.as_deref());
if let Some(user) = fallback {
let results = channels.broadcast_all(user, response).await;
for (ch, result) in results {
if let Err(e) = result {
tracing::warn!(
"Failed to broadcast heartbeat to {}: {}",
ch,
e
);
}
}
}
}
@@ -1266,7 +1286,7 @@ impl Agent {
message.channel
);
// Authorization checks (including restart channel check) are enforced in handle_system_command
self.handle_system_command(&command, &args, &message.channel)
self.handle_system_command(&command, &args, &message.channel, &message.user_id)
.await
}
Submission::Undo => self.process_undo(session, thread_id).await,
src/agent/commands.rs
+28 -13
View File
@@ -69,7 +69,7 @@ impl Agent {
}
MessageIntent::Command { command, args } => {
match self
.handle_command(&command, &args, &message.channel)
.handle_command(&command, &args, &message.channel, &message.user_id)
.await?
{
Some(s) => s,
@@ -125,6 +125,10 @@ impl Agent {
if let Some(store) = self.store()
&& let Ok(Some(ctx)) = store.get_job(uuid).await
{
// Ownership check: ensure the job belongs to the requesting user.
if ctx.user_id != user_id {
return Err(crate::error::JobError::NotFound { id: uuid }.into());
}
return Ok(format!(
"Job: {}\nStatus: {:?}\nCreated: {}\nStarted: {}\nActual cost: {}",
ctx.title,
@@ -471,6 +475,7 @@ impl Agent {
command: &str,
args: &[String],
channel: &str,
user_id: &str,
) -> Result<SubmissionResult, Error> {
match command {
"help" => Ok(SubmissionResult::response(concat!(
@@ -664,12 +669,12 @@ impl Agent {
}
if self.config.multi_tenant {
// Multi-tenant: only persist to per-user settings.
// Multi-tenant: only persist to per-user DB settings.
// Do NOT call set_model() on the shared provider — that
// would change the default for all users. The per-request
// model_override in the dispatcher reads from the same
// "selected_model" setting and applies it per-user.
self.persist_selected_model(requested).await;
self.persist_selected_model(user_id, requested).await;
Ok(SubmissionResult::response(format!(
"Model preference set to: {} (per-user)",
requested
@@ -678,7 +683,7 @@ impl Agent {
match self.llm().set_model(requested) {
Ok(()) => {
// Persist the model choice so it survives restarts.
self.persist_selected_model(requested).await;
self.persist_selected_model(user_id, requested).await;
Ok(SubmissionResult::response(format!(
"Switched model to: {}",
requested
@@ -830,10 +835,14 @@ impl Agent {
command: &str,
args: &[String],
channel: &str,
user_id: &str,
) -> Result<Option<String>, Error> {
// System commands are now handled directly via Submission::SystemCommand,
// but the router may still send us unknown /commands.
match self.handle_system_command(command, args, channel).await? {
match self
.handle_system_command(command, args, channel, user_id)
.await?
{
SubmissionResult::Response { content } => Ok(Some(content)),
SubmissionResult::Ok { message } => Ok(message),
SubmissionResult::Error { message } => Ok(Some(format!("Error: {}", message))),
@@ -845,23 +854,29 @@ impl Agent {
///
/// Best-effort: logs warnings on failure but does not propagate errors,
/// since the in-memory model switch already succeeded.
async fn persist_selected_model(&self, model: &str) {
// 1. Persist to DB if available.
///
/// In multi-tenant mode, only the per-user DB setting is written — global
/// .env and TOML files are shared across users and must not be mutated.
async fn persist_selected_model(&self, user_id: &str, model: &str) {
// 1. Persist to DB if available (per-user scoped).
if let Some(store) = self.store() {
let value = serde_json::Value::String(model.to_string());
if let Err(e) = store
.set_setting(self.owner_id(), "selected_model", &value)
.await
{
if let Err(e) = store.set_setting(user_id, "selected_model", &value).await {
tracing::warn!("Failed to persist model to DB: {}", e);
} else {
tracing::debug!("Persisted selected_model to DB: {}", model);
tracing::debug!(user_id, "Persisted selected_model to DB: {}", model);
}
} else {
tracing::warn!("No database store available — model choice will not persist to DB");
}
// 2. Update .env and TOML config file (sync I/O in spawn_blocking).
// 2. In multi-tenant mode, skip .env/TOML writes — these are global
// files shared by all users. The per-user DB setting is sufficient.
if self.config.multi_tenant {
return;
}
// 3. Update .env and TOML config file (sync I/O in spawn_blocking).
let model_owned = model.to_string();
let backend = self.deps.llm_backend.clone();
if let Err(e) = tokio::task::spawn_blocking(move || {
src/agent/heartbeat.rs
+8 -1
View File
@@ -627,7 +627,14 @@ pub fn spawn_multi_user_heartbeat(
}
// Collect results and update failure counts
while let Some(Ok((uid, result))) = join_set.join_next().await {
while let Some(join_result) = join_set.join_next().await {
let (uid, result) = match join_result {
Ok(pair) => pair,
Err(e) => {
tracing::error!("Multi-user heartbeat task panicked: {}", e);
continue;
}
};
match result {
HeartbeatResult::Ok => {
tracing::trace!(user_id = uid, "Multi-user heartbeat OK");