Skip to content

Commit 12a8a6e

Browse files
Pyinerclaude
Pyiner
and
claude
committed
fix(dispatch): re-review findings — effective run record, WS task leak, bound-agent contract
- RunRecord persists the effective run id (the active run a queued scheduled prompt merged into), so automation activity and excerpts resolve the real transcript; the requested id remains in thread logs as the dispatch correlation id. The thread-less bypass honors the same contract. - A queued chat start also aborts the caller-attached committed-stream tasks (the chat WS per-start subscriber keyed to the never-started run), not just the bound channel stream. - Thread-bound automations skip job-agent validation on create/update (the thread's agent executes; a stale automation agent must not 400 an unrelated edit) and the desktop list derives the agent pill from the bound thread — no pill when unknown — instead of showing the inert automation agent. - rustfmt: fold the run_management import block flagged in review. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1 parent f0196f8 commit 12a8a6e

5 files changed

Lines changed: 100 additions & 42 deletions

File tree

‎desktop/garyx-desktop/src/renderer/src/components/AutomationListPage.tsx‎

Lines changed: 19 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -68,9 +68,23 @@ function getTargetLabel(
6868
}
6969

7070
function getAgentLabel(
71+
state: DesktopState | null,
7172
agents: DesktopCustomAgent[],
7273
automation: DesktopAutomationSummary,
73-
): string {
74+
): string | null {
75+
const targetThreadId = automation.targetThreadId?.trim();
76+
if (targetThreadId) {
77+
// A thread-bound automation runs under the thread's own agent; the
78+
// automation-level agent does not apply. Derive the pill from the
79+
// thread, and show none rather than a wrong default when unknown.
80+
const thread = state?.threads.find((entry) => entry.id === targetThreadId);
81+
const threadAgentId = thread?.agentId?.trim();
82+
if (!threadAgentId) {
83+
return null;
84+
}
85+
const match = agents.find((agent) => agent.agentId === threadAgentId);
86+
return match?.displayName || threadAgentId;
87+
}
7488
const match = agents.find((agent) => agent.agentId === automation.agentId);
7589
return match?.displayName || automation.agentId || 'Claude';
7690
}
@@ -196,7 +210,7 @@ export function AutomationListPage({
196210
{automations.map((automation) => {
197211
const wsLabel = getWorkspaceLabel(desktopState, automation, t);
198212
const targetLabel = getTargetLabel(desktopState, automation, wsLabel, t);
199-
const agentLabel = getAgentLabel(agents, automation);
213+
const agentLabel = getAgentLabel(desktopState, agents, automation);
200214
const workspace = selectedWorkspace(desktopState, automation.workspacePath);
201215
const nextTitle = automation.schedule.kind === 'once' ? t('Run At') : t('Next');
202216
const nextRunLabel = automation.schedule.kind === 'once'
@@ -223,7 +237,9 @@ export function AutomationListPage({
223237
</div>
224238
<div style={{ display: 'flex', gap: 6, flexWrap: 'wrap', alignItems: 'center' }}>
225239
<span className={`codex-sync-pill ${status.pillClass}`}>{t(status.label)}</span>
226-
<span className="codex-sync-pill">{agentLabel}</span>
240+
{agentLabel ? (
241+
<span className="codex-sync-pill">{agentLabel}</span>
242+
) : null}
227243
<span className="codex-sync-pill ok">{formatSchedule(automation.schedule, t)}</span>
228244
{workspace && !workspace.available && (
229245
<span className="codex-sync-pill fail">{t('Workspace unavailable')}</span>

‎garyx-bridge/src/multi_provider/run_management.rs‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5,11 +5,11 @@ use std::sync::atomic::{AtomicBool, Ordering};
55
use chrono::Utc;
66
use garyx_models::provider::{
77
AgentDispatchOutcome, AgentRunRequest, FORK_FROM_PROVIDER_TYPE_METADATA_KEY,
8-
FORK_FROM_SDK_SESSION_ID_METADATA_KEY,
9-
FilePayload, ImagePayload, MODEL_METADATA_KEY, MODEL_OVERRIDE_METADATA_KEY,
10-
MODEL_REASONING_EFFORT_METADATA_KEY, MODEL_REASONING_EFFORT_OVERRIDE_METADATA_KEY,
11-
MODEL_SERVICE_TIER_METADATA_KEY, MODEL_SERVICE_TIER_OVERRIDE_METADATA_KEY, PromptAttachment,
12-
ProviderMessage, ProviderRunOptions, ProviderRunResult, ProviderType, QueuedUserInput,
8+
FORK_FROM_SDK_SESSION_ID_METADATA_KEY, FilePayload, ImagePayload, MODEL_METADATA_KEY,
9+
MODEL_OVERRIDE_METADATA_KEY, MODEL_REASONING_EFFORT_METADATA_KEY,
10+
MODEL_REASONING_EFFORT_OVERRIDE_METADATA_KEY, MODEL_SERVICE_TIER_METADATA_KEY,
11+
MODEL_SERVICE_TIER_OVERRIDE_METADATA_KEY, PromptAttachment, ProviderMessage,
12+
ProviderRunOptions, ProviderRunResult, ProviderType, QueuedUserInput,
1313
SDK_SESSION_FORK_METADATA_KEY, SDK_SESSION_ID_METADATA_KEY, StreamEvent,
1414
attachments_from_metadata, build_user_content_from_parts, stage_file_payloads_for_prompt,
1515
stage_image_payloads_for_prompt,

‎garyx-gateway/src/automation.rs‎

Lines changed: 29 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -868,15 +868,23 @@ pub async fn create_automation(
868868
Ok(value) => value,
869869
Err(error) => return invalid(error),
870870
};
871-
let agent_id = match resolve_automation_agent_id(&state, body.agent_id.as_deref(), None).await {
872-
Ok(value) => value,
873-
Err(error) => return invalid(error),
874-
};
875871
let target_thread =
876872
match resolve_automation_target_thread(&state, body.target_thread_id.as_deref()).await {
877873
Ok(value) => value,
878874
Err(error) => return invalid(error),
879875
};
876+
// A thread-bound automation executes under the thread's own agent, so
877+
// the automation-level agent is not validated (and a stale/deleted one
878+
// must not block unrelated edits). Generated-thread automations still
879+
// require a resolvable agent.
880+
let agent_id = if target_thread.is_some() {
881+
selected_agent_reference_id(body.agent_id.as_deref(), None)
882+
} else {
883+
match resolve_automation_agent_id(&state, body.agent_id.as_deref(), None).await {
884+
Ok(value) => value,
885+
Err(error) => return invalid(error),
886+
}
887+
};
880888
let explicit_workspace_dir = body
881889
.workspace_dir
882890
.as_deref()
@@ -950,16 +958,6 @@ pub async fn update_automation(
950958
},
951959
None => automation_prompt(&current),
952960
};
953-
let agent_id = match resolve_automation_agent_id(
954-
&state,
955-
body.agent_id.as_deref(),
956-
current.agent_id.as_deref(),
957-
)
958-
.await
959-
{
960-
Ok(value) => value,
961-
Err(error) => return invalid(error),
962-
};
963961
let target_thread = match &body.target_thread_id {
964962
Some(Some(thread_id)) => {
965963
match resolve_automation_target_thread(&state, Some(thread_id.as_str())).await {
@@ -973,6 +971,23 @@ pub async fn update_automation(
973971
workspace_dir: None,
974972
}),
975973
};
974+
// A thread-bound automation executes under the thread's own agent: skip
975+
// job-agent validation so a stale/deleted automation agent cannot 400 an
976+
// unrelated edit (e.g. renaming the label).
977+
let agent_id = if target_thread.is_some() {
978+
selected_agent_reference_id(body.agent_id.as_deref(), current.agent_id.as_deref())
979+
} else {
980+
match resolve_automation_agent_id(
981+
&state,
982+
body.agent_id.as_deref(),
983+
current.agent_id.as_deref(),
984+
)
985+
.await
986+
{
987+
Ok(value) => value,
988+
Err(error) => return invalid(error),
989+
}
990+
};
976991
let explicit_workspace_dir = body
977992
.workspace_dir
978993
.as_deref()

‎garyx-gateway/src/chat.rs‎

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -384,10 +384,16 @@ async fn start_chat_run(
384384
match &outcome {
385385
AgentDispatchOutcome::Started => bound_stream.detach(),
386386
AgentDispatchOutcome::QueuedToActiveRun { .. } => {
387-
// The reply belongs to the already-active run, which this
388-
// subscription (keyed to the fresh run id) will never see;
389-
// keep it alive and it leaks until process exit.
387+
// The reply belongs to the already-active run, which the
388+
// fresh-run-keyed subscriptions will never see; keeping
389+
// them alive leaks until process exit. That covers both
390+
// the bound channel stream and any caller-attached
391+
// committed-stream task (the chat WS handler's per-start
392+
// subscriber waiting on this run id).
390393
bound_stream.abort();
394+
for task in &stream_tasks {
395+
task.abort();
396+
}
391397
}
392398
}
393399
crate::runtime_diagnostics::record_message_ledger_event(

‎garyx-gateway/src/cron.rs‎

Lines changed: 38 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -1499,6 +1499,12 @@ impl CronService {
14991499

15001500
tracing::info!(job_id = %job.id, run_id = %run_id, action = ?job.action, "cron job executing");
15011501

1502+
// The run id recorded on the RunRecord. A scheduled turn queued into
1503+
// a thread's already-active run is owned by that run — automation
1504+
// activity resolves the transcript through this id, so it must be the
1505+
// effective one; the requested id stays in logs as the dispatch
1506+
// correlation id.
1507+
let mut record_run_id = run_id.clone();
15021508
let (status, error) = match &job.kind {
15031509
CronJobKind::InternalDispatch { payload } => {
15041510
// Boundary fallback: classify drop-vs-transient and
@@ -1548,7 +1554,10 @@ impl CronService {
15481554
)
15491555
.await
15501556
{
1551-
Ok(()) => (JobRunStatus::Success, None),
1557+
Ok(effective_run_id) => {
1558+
record_run_id = effective_run_id;
1559+
(JobRunStatus::Success, None)
1560+
}
15521561
Err(e) => (JobRunStatus::Failed, Some(e)),
15531562
}
15541563
}
@@ -1568,7 +1577,7 @@ impl CronService {
15681577
);
15691578

15701579
RunRecord {
1571-
run_id,
1580+
run_id: record_run_id,
15721581
job_id: job.id.clone(),
15731582
started_at,
15741583
finished_at: Some(finished_at),
@@ -1786,6 +1795,10 @@ impl CronService {
17861795
/// user message (router inbound semantics, transcript user turn, busy
17871796
/// queueing, channel echo), sharing the pipeline with `schedule_followup`
17881797
/// and the quota auto-resend instead of starting a bridge run directly.
1798+
///
1799+
/// Returns the run id that owns the reply — the requested one for a fresh
1800+
/// run, or the already-active run's id when the prompt was queued into it
1801+
/// — so run records and automation activity resolve the real transcript.
17891802
#[allow(clippy::too_many_arguments)]
17901803
async fn dispatch_agent_turn_via_thread(
17911804
job: &CronJob,
@@ -1795,7 +1808,7 @@ impl CronService {
17951808
runtime: &CronDispatchRuntime,
17961809
app_state_weak: &Arc<OnceLock<Weak<AppState>>>,
17971810
thread_key: &str,
1798-
) -> Result<(), String> {
1811+
) -> Result<String, String> {
17991812
let automation_job = is_automation_prompt_job(job);
18001813
let source = if automation_job { "automation" } else { "cron" };
18011814

@@ -1878,7 +1891,7 @@ impl CronService {
18781891
.with_field("thread_id", serde_json::json!(thread_key)),
18791892
)
18801893
.await;
1881-
Ok(())
1894+
Ok(effective_run_id)
18821895
}
18831896
Err(error) => {
18841897
runtime
@@ -1906,7 +1919,7 @@ impl CronService {
19061919
active_agent_runs: &Arc<RwLock<HashMap<String, String>>>,
19071920
dispatch_runtime: &Arc<RwLock<Option<CronDispatchRuntime>>>,
19081921
app_state_weak: &Arc<OnceLock<Weak<AppState>>>,
1909-
) -> Result<(), String> {
1922+
) -> Result<String, String> {
19101923
let runtime = dispatch_runtime
19111924
.read()
19121925
.await
@@ -2135,7 +2148,7 @@ impl CronService {
21352148
"failed to sync external user skills before scheduled dispatch"
21362149
);
21372150
}
2138-
if let Err(error) = runtime
2151+
let outcome = match runtime
21392152
.bridge
21402153
.start_agent_run(
21412154
AgentRunRequest::new(
@@ -2151,23 +2164,31 @@ impl CronService {
21512164
)
21522165
.await
21532166
{
2154-
if let Some(thread_id) = &thread_log_id {
2155-
runtime
2156-
.thread_logs
2157-
.record_event(
2158-
ThreadLogEvent::error(thread_id, "automation", "scheduled dispatch failed")
2167+
Ok(outcome) => outcome,
2168+
Err(error) => {
2169+
if let Some(thread_id) = &thread_log_id {
2170+
runtime
2171+
.thread_logs
2172+
.record_event(
2173+
ThreadLogEvent::error(
2174+
thread_id,
2175+
"automation",
2176+
"scheduled dispatch failed",
2177+
)
21592178
.with_run_id(run_id.to_owned())
21602179
.with_field("job_id", serde_json::json!(job.id))
21612180
.with_field("error", serde_json::json!(error.to_string())),
2162-
)
2163-
.await;
2181+
)
2182+
.await;
2183+
}
2184+
return Err(format!("cron dispatch failed: {error}"));
21642185
}
2165-
return Err(format!("cron dispatch failed: {error}"));
2166-
}
2186+
};
2187+
let effective_run_id = outcome.effective_run_id().unwrap_or(run_id).to_owned();
21672188
active_agent_runs
21682189
.write()
21692190
.await
2170-
.insert(job.id.clone(), run_id.to_owned());
2191+
.insert(job.id.clone(), effective_run_id.clone());
21712192
if let Some(thread_id) = &thread_log_id {
21722193
runtime
21732194
.thread_logs
@@ -2180,7 +2201,7 @@ impl CronService {
21802201
.await;
21812202
}
21822203

2183-
Ok(())
2204+
Ok(effective_run_id)
21842205
}
21852206
}
21862207

0 commit comments

Comments
 (0)