Skip to content
This repository was archived by the owner on Apr 11, 2026. It is now read-only.

Commit edd3648

Browse files
z23ccclaude
z23cc
and
claude
committed
feat: Wave 2 — WebSocket protocol + event_bus emit
- 5 new FlowEvent variants: TaskStatusChanged, DagMutated, AgentLog, EpicUpdated, Heartbeat - All POST handlers emit events after successful mutations - Heartbeat every 30s for WebSocket keep-alive - 292 tests pass, clippy clean Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1 parent 9fd731b commit edd3648

5 files changed

Lines changed: 231 additions & 43 deletions

File tree

‎flowctl/crates/flowctl-daemon/src/handlers/dag.rs‎

Lines changed: 26 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,8 @@ use axum::Json;
66

77
use flowctl_core::state_machine::{Status, Transition};
88

9+
use flowctl_scheduler::FlowEvent;
10+
911
use super::common::{check_version, touch_updated_at, AppError, AppState};
1012

1113
/// A node in the DAG visualization.
@@ -211,9 +213,9 @@ pub async fn dag_mutate_handler(
211213
).map_err(|e| AppError::Db(e.to_string()))?;
212214
touch_updated_at(&conn, task_id)?;
213215

214-
state.event_bus.emit(flowctl_scheduler::FlowEvent::TaskReady {
215-
task_id: task_id.to_string(),
216-
epic_id: task.epic.clone(),
216+
state.event_bus.emit(FlowEvent::DagMutated {
217+
mutation: "dep_added".to_string(),
218+
details: serde_json::json!({"from": depends_on, "to": task_id}),
217219
});
218220

219221
Ok(Json(serde_json::json!({"success": true, "action": "add_dep"})))
@@ -237,9 +239,9 @@ pub async fn dag_mutate_handler(
237239
).map_err(|e| AppError::Db(e.to_string()))?;
238240
touch_updated_at(&conn, task_id)?;
239241

240-
state.event_bus.emit(flowctl_scheduler::FlowEvent::TaskReady {
241-
task_id: task_id.to_string(),
242-
epic_id: task.epic.clone(),
242+
state.event_bus.emit(FlowEvent::DagMutated {
243+
mutation: "dep_removed".to_string(),
244+
details: serde_json::json!({"from": depends_on, "to": task_id}),
243245
});
244246

245247
Ok(Json(serde_json::json!({"success": true, "action": "remove_dep"})))
@@ -258,11 +260,14 @@ pub async fn dag_mutate_handler(
258260
AppError::InvalidTransition(format!("cannot retry task '{}': {}", task_id, e))
259261
})?;
260262

263+
let from_status = format!("{:?}", task.status).to_lowercase();
261264
repo.update_status(task_id, Status::Todo)?;
262265

263-
state.event_bus.emit(flowctl_scheduler::FlowEvent::TaskReady {
266+
state.event_bus.emit(FlowEvent::TaskStatusChanged {
264267
task_id: task_id.to_string(),
265268
epic_id: task.epic.clone(),
269+
from_status,
270+
to_status: "todo".to_string(),
266271
});
267272

268273
Ok(Json(serde_json::json!({"success": true, "action": "retry_task"})))
@@ -281,11 +286,14 @@ pub async fn dag_mutate_handler(
281286
AppError::InvalidTransition(format!("cannot skip task '{}': {}", task_id, e))
282287
})?;
283288

289+
let from_status = format!("{:?}", task.status).to_lowercase();
284290
repo.update_status(task_id, Status::Skipped)?;
285291

286-
state.event_bus.emit(flowctl_scheduler::FlowEvent::TaskReady {
292+
state.event_bus.emit(FlowEvent::TaskStatusChanged {
287293
task_id: task_id.to_string(),
288294
epic_id: task.epic.clone(),
295+
from_status,
296+
to_status: "skipped".to_string(),
289297
});
290298

291299
Ok(Json(serde_json::json!({"success": true, "action": "skip_task"})))
@@ -327,6 +335,11 @@ pub async fn add_dep_handler(
327335
).map_err(|e| AppError::Db(e.to_string()))?;
328336
touch_updated_at(&conn, &body.to)?;
329337

338+
state.event_bus.emit(FlowEvent::DagMutated {
339+
mutation: "dep_added".to_string(),
340+
details: serde_json::json!({"from": body.from, "to": body.to}),
341+
});
342+
330343
Ok((
331344
StatusCode::CREATED,
332345
Json(serde_json::json!({"from": body.from, "to": body.to})),
@@ -353,6 +366,11 @@ pub async fn remove_dep_handler(
353366

354367
touch_updated_at(&conn, &to)?;
355368

369+
state.event_bus.emit(FlowEvent::DagMutated {
370+
mutation: "dep_removed".to_string(),
371+
details: serde_json::json!({"from": from, "to": to}),
372+
});
373+
356374
Ok(Json(serde_json::json!({"from": from, "to": to})))
357375
}
358376

‎flowctl/crates/flowctl-daemon/src/handlers/epic.rs‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,8 @@ use axum::Json;
66

77
use flowctl_core::id::slugify;
88

9+
use flowctl_scheduler::FlowEvent;
10+
911
use super::common::{AppError, AppState};
1012

1113
/// POST /api/v1/epics/create -- create a new epic.
@@ -95,6 +97,12 @@ pub async fn set_epic_plan_handler(
9597
rusqlite::params![chrono::Utc::now().to_rfc3339(), epic_id],
9698
).map_err(|e| AppError::Db(e.to_string()))?;
9799

100+
state.event_bus.emit(FlowEvent::EpicUpdated {
101+
epic_id: epic_id.clone(),
102+
field: "plan".to_string(),
103+
value: serde_json::Value::String("updated".to_string()),
104+
});
105+
98106
Ok(Json(serde_json::json!({"id": epic_id})))
99107
}
100108

@@ -139,6 +147,12 @@ pub async fn start_epic_work_handler(
139147
}
140148
}
141149

150+
state.event_bus.emit(FlowEvent::EpicUpdated {
151+
epic_id: epic_id.clone(),
152+
field: "work_started".to_string(),
153+
value: serde_json::json!({"tasks_started": tasks_started}),
154+
});
155+
142156
Ok(Json(serde_json::json!({
143157
"id": epic_id,
144158
"tasks_started": tasks_started

‎flowctl/crates/flowctl-daemon/src/handlers/task.rs‎

Lines changed: 125 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -4,9 +4,10 @@ use axum::extract::State;
44
use axum::http::StatusCode;
55
use axum::Json;
66

7-
use flowctl_core::id::is_task_id;
7+
use flowctl_core::id::{epic_id_from_task, is_task_id};
88
use flowctl_core::state_machine::{Status, Transition};
99
use flowctl_core::types::FLOW_DIR;
10+
use flowctl_scheduler::FlowEvent;
1011
use flowctl_service::lifecycle::{BlockTaskRequest, DoneTaskRequest, RestartTaskRequest, StartTaskRequest};
1112
use flowctl_service::ServiceError;
1213

@@ -77,9 +78,18 @@ pub async fn start_task_handler(
7778
.map_err(|e| AppError::Internal(format!("spawn_blocking failed: {e}")))?;
7879

7980
match result {
80-
Ok(resp) => Ok(Json(
81-
serde_json::json!({"success": true, "id": resp.task_id}),
82-
)),
81+
Ok(resp) => {
82+
let epic_id = epic_id_from_task(&resp.task_id).unwrap_or_default();
83+
state.event_bus.emit(FlowEvent::TaskStatusChanged {
84+
task_id: resp.task_id.clone(),
85+
epic_id,
86+
from_status: "todo".to_string(),
87+
to_status: format!("{:?}", resp.status).to_lowercase(),
88+
});
89+
Ok(Json(
90+
serde_json::json!({"success": true, "id": resp.task_id}),
91+
))
92+
}
8393
Err(e) => Err(service_error_to_app_error(e)),
8494
}
8595
}
@@ -109,10 +119,19 @@ pub async fn start_task_rest_handler(
109119
.map_err(|e| AppError::Internal(format!("spawn_blocking failed: {e}")))?;
110120

111121
match result {
112-
Ok(resp) => Ok(Json(serde_json::json!({
113-
"id": resp.task_id,
114-
"status": format!("{:?}", resp.status).to_lowercase()
115-
}))),
122+
Ok(resp) => {
123+
let epic_id = epic_id_from_task(&resp.task_id).unwrap_or_default();
124+
state.event_bus.emit(FlowEvent::TaskStatusChanged {
125+
task_id: resp.task_id.clone(),
126+
epic_id,
127+
from_status: "todo".to_string(),
128+
to_status: format!("{:?}", resp.status).to_lowercase(),
129+
});
130+
Ok(Json(serde_json::json!({
131+
"id": resp.task_id,
132+
"status": format!("{:?}", resp.status).to_lowercase()
133+
})))
134+
}
116135
Err(e) => Err(service_error_to_app_error(e)),
117136
}
118137
}
@@ -148,11 +167,20 @@ pub async fn done_task_rest_handler(
148167
.map_err(|e| AppError::Internal(format!("spawn_blocking failed: {e}")))?;
149168

150169
match result {
151-
Ok(resp) => Ok(Json(serde_json::json!({
152-
"id": resp.task_id,
153-
"status": format!("{:?}", resp.status).to_lowercase(),
154-
"duration_seconds": resp.duration_seconds
155-
}))),
170+
Ok(resp) => {
171+
let epic_id = epic_id_from_task(&resp.task_id).unwrap_or_default();
172+
state.event_bus.emit(FlowEvent::TaskStatusChanged {
173+
task_id: resp.task_id.clone(),
174+
epic_id,
175+
from_status: "in_progress".to_string(),
176+
to_status: format!("{:?}", resp.status).to_lowercase(),
177+
});
178+
Ok(Json(serde_json::json!({
179+
"id": resp.task_id,
180+
"status": format!("{:?}", resp.status).to_lowercase(),
181+
"duration_seconds": resp.duration_seconds
182+
})))
183+
}
156184
Err(e) => Err(service_error_to_app_error(e)),
157185
}
158186
}
@@ -180,10 +208,19 @@ pub async fn block_task_rest_handler(
180208
.map_err(|e| AppError::Internal(format!("spawn_blocking failed: {e}")))?;
181209

182210
match result {
183-
Ok(resp) => Ok(Json(serde_json::json!({
184-
"id": resp.task_id,
185-
"status": format!("{:?}", resp.status).to_lowercase()
186-
}))),
211+
Ok(resp) => {
212+
let epic_id = epic_id_from_task(&resp.task_id).unwrap_or_default();
213+
state.event_bus.emit(FlowEvent::TaskStatusChanged {
214+
task_id: resp.task_id.clone(),
215+
epic_id,
216+
from_status: "in_progress".to_string(),
217+
to_status: format!("{:?}", resp.status).to_lowercase(),
218+
});
219+
Ok(Json(serde_json::json!({
220+
"id": resp.task_id,
221+
"status": format!("{:?}", resp.status).to_lowercase()
222+
})))
223+
}
187224
Err(e) => Err(service_error_to_app_error(e)),
188225
}
189226
}
@@ -216,11 +253,20 @@ pub async fn restart_task_rest_handler(
216253
.map_err(|e| AppError::Internal(format!("spawn_blocking failed: {e}")))?;
217254

218255
match result {
219-
Ok(resp) => Ok(Json(serde_json::json!({
220-
"id": resp.cascade_from,
221-
"status": "todo",
222-
"cascaded": resp.reset_ids
223-
}))),
256+
Ok(resp) => {
257+
let epic_id = epic_id_from_task(&resp.cascade_from).unwrap_or_default();
258+
state.event_bus.emit(FlowEvent::TaskStatusChanged {
259+
task_id: resp.cascade_from.clone(),
260+
epic_id,
261+
from_status: "done".to_string(),
262+
to_status: "todo".to_string(),
263+
});
264+
Ok(Json(serde_json::json!({
265+
"id": resp.cascade_from,
266+
"status": "todo",
267+
"cascaded": resp.reset_ids
268+
})))
269+
}
224270
Err(e) => Err(service_error_to_app_error(e)),
225271
}
226272
}
@@ -237,12 +283,21 @@ pub async fn skip_task_rest_handler(
237283
.get(&task_id)
238284
.map_err(|_| AppError::InvalidInput(format!("task not found: {task_id}")))?;
239285

286+
let from_status = format!("{:?}", task.status).to_lowercase();
240287
Transition::new(task.status, Status::Skipped).map_err(|e| {
241288
AppError::InvalidTransition(format!("cannot skip task '{task_id}': {e}"))
242289
})?;
243290

244291
repo.update_status(&task_id, Status::Skipped)?;
245292

293+
let epic_id = epic_id_from_task(&task_id).unwrap_or_default();
294+
state.event_bus.emit(FlowEvent::TaskStatusChanged {
295+
task_id: task_id.clone(),
296+
epic_id,
297+
from_status,
298+
to_status: "skipped".to_string(),
299+
});
300+
246301
Ok(Json(serde_json::json!({
247302
"id": task_id,
248303
"status": "skipped"
@@ -280,9 +335,18 @@ pub async fn done_task_handler(
280335
.map_err(|e| AppError::Internal(format!("spawn_blocking failed: {e}")))?;
281336

282337
match result {
283-
Ok(resp) => Ok(Json(
284-
serde_json::json!({"success": true, "id": resp.task_id}),
285-
)),
338+
Ok(resp) => {
339+
let epic_id = epic_id_from_task(&resp.task_id).unwrap_or_default();
340+
state.event_bus.emit(FlowEvent::TaskStatusChanged {
341+
task_id: resp.task_id.clone(),
342+
epic_id,
343+
from_status: "in_progress".to_string(),
344+
to_status: format!("{:?}", resp.status).to_lowercase(),
345+
});
346+
Ok(Json(
347+
serde_json::json!({"success": true, "id": resp.task_id}),
348+
))
349+
}
286350
Err(e) => Err(service_error_to_app_error(e)),
287351
}
288352
}
@@ -298,12 +362,21 @@ pub async fn skip_task_handler(
298362
.get(&body.task_id)
299363
.map_err(|_| AppError::InvalidInput(format!("task not found: {}", body.task_id)))?;
300364

365+
let from_status = format!("{:?}", task.status).to_lowercase();
301366
Transition::new(task.status, Status::Skipped).map_err(|e| {
302367
AppError::InvalidTransition(format!("cannot skip task '{}': {}", body.task_id, e))
303368
})?;
304369

305370
repo.update_status(&body.task_id, Status::Skipped)?;
306371

372+
let epic_id = epic_id_from_task(&body.task_id).unwrap_or_default();
373+
state.event_bus.emit(FlowEvent::TaskStatusChanged {
374+
task_id: body.task_id.clone(),
375+
epic_id,
376+
from_status,
377+
to_status: "skipped".to_string(),
378+
});
379+
307380
Ok(Json(serde_json::json!({
308381
"success": true,
309382
"id": body.task_id,
@@ -334,11 +407,20 @@ pub async fn block_task_handler(
334407
.map_err(|e| AppError::Internal(format!("spawn_blocking failed: {e}")))?;
335408

336409
match result {
337-
Ok(resp) => Ok(Json(serde_json::json!({
338-
"success": true,
339-
"id": resp.task_id,
340-
"status": "blocked"
341-
}))),
410+
Ok(resp) => {
411+
let epic_id = epic_id_from_task(&resp.task_id).unwrap_or_default();
412+
state.event_bus.emit(FlowEvent::TaskStatusChanged {
413+
task_id: resp.task_id.clone(),
414+
epic_id,
415+
from_status: "in_progress".to_string(),
416+
to_status: format!("{:?}", resp.status).to_lowercase(),
417+
});
418+
Ok(Json(serde_json::json!({
419+
"success": true,
420+
"id": resp.task_id,
421+
"status": "blocked"
422+
})))
423+
}
342424
Err(e) => Err(service_error_to_app_error(e)),
343425
}
344426
}
@@ -369,10 +451,19 @@ pub async fn restart_task_handler(
369451
.map_err(|e| AppError::Internal(format!("spawn_blocking failed: {e}")))?;
370452

371453
match result {
372-
Ok(resp) => Ok(Json(serde_json::json!({
373-
"success": true,
374-
"reset": resp.reset_ids
375-
}))),
454+
Ok(resp) => {
455+
let epic_id = epic_id_from_task(&resp.cascade_from).unwrap_or_default();
456+
state.event_bus.emit(FlowEvent::TaskStatusChanged {
457+
task_id: resp.cascade_from.clone(),
458+
epic_id,
459+
from_status: "done".to_string(),
460+
to_status: "todo".to_string(),
461+
});
462+
Ok(Json(serde_json::json!({
463+
"success": true,
464+
"reset": resp.reset_ids
465+
})))
466+
}
376467
Err(e) => Err(service_error_to_app_error(e)),
377468
}
378469
}

0 commit comments

Comments
 (0)