Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/agent/channel.rs
Original file line number Diff line number Diff line change
Expand Up @@ -164,7 +164,7 @@ impl ChannelState {
};

self.process_run_logger
.log_worker_completed(worker_id, &result, false);
.log_worker_cancelled(worker_id, &result);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice to persist cancelled explicitly. One thing to keep an eye on: the synthetic WorkerComplete { success: false } is still indistinguishable from a real failure for any downstream consumers that don't consult worker_runs.status (or parse the result text). If anything still maps success=false => failed, it may be worth threading a terminal status through the event shape in a follow-up.

if let Err(error) = self.deps.event_tx.send(ProcessEvent::WorkerComplete {
agent_id: self.deps.agent_id.clone(),
worker_id,
Expand Down
88 changes: 87 additions & 1 deletion src/conversation/history.rs
Original file line number Diff line number Diff line change
Expand Up @@ -475,10 +475,20 @@ impl ProcessRunLogger {

/// Record a worker completing with its result. Fire-and-forget.
pub fn log_worker_completed(&self, worker_id: WorkerId, result: &str, success: bool) {
let status = if success { "done" } else { "failed" };
self.log_worker_completed_with_status(worker_id, result, status);
}

/// Record a worker as cancelled. Fire-and-forget.
pub fn log_worker_cancelled(&self, worker_id: WorkerId, result: &str) {
self.log_worker_completed_with_status(worker_id, result, "cancelled");
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved

fn log_worker_completed_with_status(&self, worker_id: WorkerId, result: &str, status: &str) {
let pool = self.pool.clone();
let id = worker_id.to_string();
let result = result.to_string();
let status = if success { "done" } else { "failed" };
let status = status.to_string();

tokio::spawn(async move {
if let Err(error) = sqlx::query(

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since log_worker_started() is also fire-and-forget, it seems possible for this UPDATE to race ahead of the INSERT on very fast cancellations and silently affect 0 rows. You already handle this pattern in log_worker_project_link() / log_opencode_metadata().

Might be worth adding a small retry on rows_affected() == 0 here so we don't lose the terminal state, and guarding against overwriting an already-terminal row.

Suggested change
if let Err(error) = sqlx::query(
"UPDATE worker_runs SET result = ?, status = ?, completed_at = CURRENT_TIMESTAMP WHERE id = ? AND completed_at IS NULL"

Expand Down Expand Up @@ -1054,6 +1064,82 @@ mod tests {
assert_eq!(result, "Worker cancelled");
}

#[tokio::test]
async fn log_worker_completed_success_sets_done_status() {
let pool = setup_worker_runs_table().await;
let logger = ProcessRunLogger::new(pool.clone());
let worker_id = uuid::Uuid::new_v4();

sqlx::query("INSERT INTO worker_runs (id, channel_id, status, result) VALUES (?, 'ch-1', 'running', '')")
.bind(worker_id.to_string())
.execute(&pool)
.await
.expect("insert");

logger.log_worker_completed(worker_id, "task finished", true);
tokio::time::sleep(std::time::Duration::from_millis(50)).await;

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This sleep(50ms) makes the test dependent on timing for the fire-and-forget tokio::spawn, which can get flaky on slow CI. Might be worth polling worker_runs until the status transitions (with a short timeout) instead of using a fixed sleep.


let row = sqlx::query("SELECT status, result FROM worker_runs WHERE id = ?")
.bind(worker_id.to_string())
.fetch_one(&pool)
.await
.expect("fetch");
let status: String = sqlx::Row::try_get(&row, "status").expect("status");
let result: String = sqlx::Row::try_get(&row, "result").expect("result");
assert_eq!(status, "done");
assert_eq!(result, "task finished");
}

#[tokio::test]
async fn log_worker_completed_failure_sets_failed_status() {
let pool = setup_worker_runs_table().await;
let logger = ProcessRunLogger::new(pool.clone());
let worker_id = uuid::Uuid::new_v4();

sqlx::query("INSERT INTO worker_runs (id, channel_id, status, result) VALUES (?, 'ch-1', 'running', '')")
.bind(worker_id.to_string())
.execute(&pool)
.await
.expect("insert");

logger.log_worker_completed(worker_id, "something broke", false);
tokio::time::sleep(std::time::Duration::from_millis(50)).await;

let row = sqlx::query("SELECT status FROM worker_runs WHERE id = ?")
.bind(worker_id.to_string())
.fetch_one(&pool)
.await
.expect("fetch");
let status: String = sqlx::Row::try_get(&row, "status").expect("status");
assert_eq!(status, "failed");
}

#[tokio::test]
async fn log_worker_cancelled_sets_cancelled_status() {
let pool = setup_worker_runs_table().await;
let logger = ProcessRunLogger::new(pool.clone());
let worker_id = uuid::Uuid::new_v4();

sqlx::query("INSERT INTO worker_runs (id, channel_id, status, result) VALUES (?, 'ch-1', 'running', '')")
.bind(worker_id.to_string())
.execute(&pool)
.await
.expect("insert");

logger.log_worker_cancelled(worker_id, "Worker cancelled: user requested");
tokio::time::sleep(std::time::Duration::from_millis(50)).await;

let row = sqlx::query("SELECT status, result FROM worker_runs WHERE id = ?")
.bind(worker_id.to_string())
.fetch_one(&pool)
.await
.expect("fetch");
let status: String = sqlx::Row::try_get(&row, "status").expect("status");
let result: String = sqlx::Row::try_get(&row, "result").expect("result");
assert_eq!(status, "cancelled");
assert_eq!(result, "Worker cancelled: user requested");
}

#[tokio::test]
async fn cancel_running_detached_worker_does_not_touch_channel_bound_rows() {
let pool = setup_worker_runs_table().await;
Expand Down
54 changes: 25 additions & 29 deletions src/tools/send_message_to_another_channel.rs
Original file line number Diff line number Diff line change
Expand Up @@ -150,14 +150,13 @@ impl Tool for SendMessageTool {
// If explicit prefix returned default "signal" adapter but we're in a named
// Signal adapter conversation (e.g., signal:gvoice1), use the current adapter
// to ensure the message goes through the correct account.
if target.adapter == "signal" {
if let Some(current_adapter) = self
if target.adapter == "signal"
&& let Some(current_adapter) = self
.current_adapter
.as_ref()
.filter(|adapter| adapter.starts_with("signal:"))
{
target.adapter = current_adapter.clone();
}
{
target.adapter = current_adapter.clone();
}

self.messaging_manager
Expand Down Expand Up @@ -189,31 +188,28 @@ impl Tool for SendMessageTool {
.current_adapter
.as_ref()
.filter(|adapter| adapter.starts_with("signal"))
&& let Some(target) = parse_implicit_signal_shorthand(&args.target, current_adapter)
{
if let Some(target) = parse_implicit_signal_shorthand(&args.target, current_adapter) {
self.messaging_manager
.broadcast(
&target.adapter,
&target.target,
crate::OutboundResponse::Text(args.message),
)
.await
.map_err(|error| {
SendMessageError(format!("failed to send message: {error}"))
})?;

tracing::info!(
adapter = %target.adapter,
broadcast_target = %"[REDACTED]",
"message sent via implicit Signal shorthand"
);

return Ok(SendMessageOutput {
success: true,
target: target.target,
platform: target.adapter,
});
}
self.messaging_manager
.broadcast(
&target.adapter,
&target.target,
crate::OutboundResponse::Text(args.message),
)
.await
.map_err(|error| SendMessageError(format!("failed to send message: {error}")))?;

tracing::info!(
adapter = %target.adapter,
broadcast_target = %"[REDACTED]",
"message sent via implicit Signal shorthand"
);

return Ok(SendMessageOutput {
success: true,
target: target.target,
platform: target.adapter,
});
}

// Check for explicit email target
Expand Down
Loading