@@ -73,9 +73,9 @@ mod imp {
7373 QueueReorderRequest , QueueStats , SubjectCreateRequest , SubjectGetRequest ,
7474 SubjectListRequest , SubjectListResponse , SubjectNextRequest , SubjectNextResponse ,
7575 SubjectStatusRequest , SubjectUpdateRequest , SubjectWatchRequest , Unit ,
76- WorkflowCancelRequest , WorkflowExecuteRequest , WorkflowGetRequest , WorkflowListRequest ,
77- WorkflowListResponse , WorkflowPauseRequest , WorkflowResumeRequest , WorkflowRun ,
78- WorkflowRunRequest , WorkflowRunStart ,
76+ WorkflowCancelRequest , WorkflowEvent , WorkflowEventsRequest , WorkflowExecuteRequest ,
77+ WorkflowGetRequest , WorkflowListRequest , WorkflowListResponse , WorkflowPauseRequest ,
78+ WorkflowResumeRequest , WorkflowRun , WorkflowRunRequest , WorkflowRunStart ,
7979 } ;
8080 use animus_subject_protocol:: { Subject , SubjectChangedEvent } ;
8181
@@ -417,6 +417,23 @@ mod imp {
417417 self . rpc ( method:: METHOD_WORKFLOW_CANCEL , request) . await
418418 }
419419
420+ /// Open a `workflow/events` subscription (v0.1.10).
421+ ///
422+ /// Returns a [`Subscription`] yielding [`WorkflowEvent`] items as the
423+ /// daemon emits workflow-scoped events. The request optionally filters
424+ /// by `workflow_id` and/or event `kinds`; see [`WorkflowEventsRequest`].
425+ ///
426+ /// NOTE: daemons that have not yet implemented `workflow/events` will
427+ /// return a `method_not_found` error on subscribe. Clients SHOULD
428+ /// degrade to `daemon_events` + kind filtering in that case.
429+ pub async fn workflow_events (
430+ & self ,
431+ request : WorkflowEventsRequest ,
432+ ) -> Result < Subscription < WorkflowEvent > > {
433+ self . subscribe ( method:: METHOD_WORKFLOW_EVENTS , request, 256 )
434+ . await
435+ }
436+
420437 // ---- Queue --------------------------------------------------------
421438
422439 /// Call `queue/list`.
@@ -1009,6 +1026,173 @@ mod imp {
10091026 ) ;
10101027 }
10111028
1029+ #[ tokio:: test]
1030+ async fn workflow_events_streams_events_and_cancels_on_drop ( ) {
1031+ let tmp = TempDir :: new ( ) . unwrap ( ) ;
1032+ let socket = tmp. path ( ) . join ( "control.sock" ) ;
1033+ let listener = UnixListener :: bind ( & socket) . unwrap ( ) ;
1034+ let socket_path = socket. clone ( ) ;
1035+
1036+ let received_cancel = Arc :: new ( AsyncMutex :: new ( false ) ) ;
1037+ let received_cancel_clone = Arc :: clone ( & received_cancel) ;
1038+
1039+ tokio:: spawn ( async move {
1040+ let ( conn, _) = listener. accept ( ) . await . unwrap ( ) ;
1041+ let ( read_half, write_half) = conn. into_split ( ) ;
1042+ let write_half = Arc :: new ( AsyncMutex :: new ( write_half) ) ;
1043+ let mut reader = BufReader :: new ( read_half) ;
1044+
1045+ let mut line = String :: new ( ) ;
1046+ reader. read_line ( & mut line) . await . unwrap ( ) ;
1047+ let req: RpcRequest = serde_json:: from_str ( line. trim ( ) ) . unwrap ( ) ;
1048+ assert_eq ! ( req. method, method:: METHOD_WORKFLOW_EVENTS ) ;
1049+ let req_id = req. id . clone ( ) ;
1050+
1051+ let ack = RpcResponse {
1052+ jsonrpc : "2.0" . into ( ) ,
1053+ id : req_id. clone ( ) ,
1054+ result : Some ( serde_json:: json!( { "watching" : true } ) ) ,
1055+ error : None ,
1056+ } ;
1057+ {
1058+ let mut g = write_half. lock ( ) . await ;
1059+ let mut frame = serde_json:: to_vec ( & ack) . unwrap ( ) ;
1060+ frame. push ( b'\n' ) ;
1061+ g. write_all ( & frame) . await . unwrap ( ) ;
1062+ g. flush ( ) . await . unwrap ( ) ;
1063+ }
1064+
1065+ for i in 0 ..3u64 {
1066+ let kind = if i % 2 == 0 {
1067+ "phase_started"
1068+ } else {
1069+ "phase_completed"
1070+ } ;
1071+ let event = serde_json:: json!( {
1072+ "workflow_id" : "wf-1" ,
1073+ "kind" : kind,
1074+ "payload" : { "seq" : i } ,
1075+ "occurred_at" : Utc :: now( ) . to_rfc3339( ) ,
1076+ } ) ;
1077+ let notification = RpcNotification :: new (
1078+ method:: NOTIFICATION_WORKFLOW_EVENT . to_string ( ) ,
1079+ Some ( serde_json:: json!( {
1080+ "id" : req_id,
1081+ "data" : event,
1082+ } ) ) ,
1083+ ) ;
1084+ let mut g = write_half. lock ( ) . await ;
1085+ let mut frame = serde_json:: to_vec ( & notification) . unwrap ( ) ;
1086+ frame. push ( b'\n' ) ;
1087+ g. write_all ( & frame) . await . unwrap ( ) ;
1088+ g. flush ( ) . await . unwrap ( ) ;
1089+ drop ( g) ;
1090+ tokio:: time:: sleep ( Duration :: from_millis ( 5 ) ) . await ;
1091+ }
1092+
1093+ let mut cancel_line = String :: new ( ) ;
1094+ if timeout ( Duration :: from_secs ( 2 ) , reader. read_line ( & mut cancel_line) )
1095+ . await
1096+ . is_ok ( )
1097+ {
1098+ if let Ok ( v) = serde_json:: from_str :: < Value > ( cancel_line. trim ( ) ) {
1099+ if v. get ( "method" ) . and_then ( |m| m. as_str ( ) ) == Some ( "$/cancelRequest" ) {
1100+ * received_cancel_clone. lock ( ) . await = true ;
1101+ }
1102+ }
1103+ }
1104+ } ) ;
1105+
1106+ let client = ControlClient :: connect ( & socket_path) . await . unwrap ( ) ;
1107+ let mut sub = client
1108+ . workflow_events ( WorkflowEventsRequest :: default ( ) )
1109+ . await
1110+ . unwrap ( ) ;
1111+
1112+ for expected in 0 ..3u64 {
1113+ let ev = timeout ( Duration :: from_secs ( 2 ) , sub. recv ( ) )
1114+ . await
1115+ . expect ( "recv timeout" )
1116+ . expect ( "stream closed early" ) ;
1117+ assert_eq ! ( ev. workflow_id, "wf-1" ) ;
1118+ assert_eq ! (
1119+ ev. payload. get( "seq" ) . and_then( |v| v. as_u64( ) ) ,
1120+ Some ( expected)
1121+ ) ;
1122+ let expected_kind = if expected % 2 == 0 {
1123+ "phase_started"
1124+ } else {
1125+ "phase_completed"
1126+ } ;
1127+ assert_eq ! ( ev. kind, expected_kind) ;
1128+ }
1129+
1130+ drop ( sub) ;
1131+
1132+ tokio:: time:: sleep ( Duration :: from_millis ( 200 ) ) . await ;
1133+ assert ! (
1134+ * received_cancel. lock( ) . await ,
1135+ "server did not see $/cancelRequest"
1136+ ) ;
1137+ }
1138+
1139+ #[ tokio:: test]
1140+ async fn workflow_events_filters_by_workflow_id_in_request ( ) {
1141+ let tmp = TempDir :: new ( ) . unwrap ( ) ;
1142+ let socket = tmp. path ( ) . join ( "control.sock" ) ;
1143+ let listener = UnixListener :: bind ( & socket) . unwrap ( ) ;
1144+ let socket_path = socket. clone ( ) ;
1145+
1146+ let captured_params: Arc < AsyncMutex < Option < Value > > > = Arc :: new ( AsyncMutex :: new ( None ) ) ;
1147+ let captured_clone = Arc :: clone ( & captured_params) ;
1148+
1149+ tokio:: spawn ( async move {
1150+ let ( conn, _) = listener. accept ( ) . await . unwrap ( ) ;
1151+ let ( read_half, mut write_half) = conn. into_split ( ) ;
1152+ let mut reader = BufReader :: new ( read_half) ;
1153+
1154+ let mut line = String :: new ( ) ;
1155+ reader. read_line ( & mut line) . await . unwrap ( ) ;
1156+ let req: RpcRequest = serde_json:: from_str ( line. trim ( ) ) . unwrap ( ) ;
1157+ assert_eq ! ( req. method, method:: METHOD_WORKFLOW_EVENTS ) ;
1158+ * captured_clone. lock ( ) . await = req. params . clone ( ) ;
1159+
1160+ let ack = RpcResponse {
1161+ jsonrpc : "2.0" . into ( ) ,
1162+ id : req. id . clone ( ) ,
1163+ result : Some ( serde_json:: json!( { "watching" : true } ) ) ,
1164+ error : None ,
1165+ } ;
1166+ let mut frame = serde_json:: to_vec ( & ack) . unwrap ( ) ;
1167+ frame. push ( b'\n' ) ;
1168+ write_half. write_all ( & frame) . await . unwrap ( ) ;
1169+ write_half. flush ( ) . await . unwrap ( ) ;
1170+ tokio:: time:: sleep ( Duration :: from_millis ( 50 ) ) . await ;
1171+ } ) ;
1172+
1173+ let client = ControlClient :: connect ( & socket_path) . await . unwrap ( ) ;
1174+ let _sub = client
1175+ . workflow_events ( WorkflowEventsRequest {
1176+ workflow_id : Some ( "wf-42" . into ( ) ) ,
1177+ kinds : Some ( vec ! [ "phase_completed" . into( ) , "workflow_failed" . into( ) ] ) ,
1178+ } )
1179+ . await
1180+ . unwrap ( ) ;
1181+
1182+ tokio:: time:: sleep ( Duration :: from_millis ( 50 ) ) . await ;
1183+ let params = captured_params. lock ( ) . await . clone ( ) . expect ( "no params" ) ;
1184+ assert_eq ! (
1185+ params. get( "workflow_id" ) . and_then( |v| v. as_str( ) ) ,
1186+ Some ( "wf-42" )
1187+ ) ;
1188+ let kinds = params
1189+ . get ( "kinds" )
1190+ . and_then ( |v| v. as_array ( ) )
1191+ . expect ( "kinds not an array" ) ;
1192+ let kinds: Vec < & str > = kinds. iter ( ) . filter_map ( |v| v. as_str ( ) ) . collect ( ) ;
1193+ assert_eq ! ( kinds, vec![ "phase_completed" , "workflow_failed" ] ) ;
1194+ }
1195+
10121196 #[ tokio:: test]
10131197 async fn concurrent_rpc_and_subscriptions ( ) {
10141198 let tmp = TempDir :: new ( ) . unwrap ( ) ;
0 commit comments