@@ -5,22 +5,26 @@ import {
55 CommandId ,
66 EventId ,
77 GuestShareId ,
8+ type GuestStreamEvent ,
89 MessageId ,
910 type OrchestrationV2TurnItem ,
1011 ProjectId ,
1112 ProviderDriverKind ,
1213 ProviderInstanceId ,
14+ RunId ,
1315 ThreadId ,
1416 TurnItemId ,
1517} from "@t3tools/contracts" ;
1618import * as Context from "effect/Context" ;
1719import * as DateTime from "effect/DateTime" ;
20+ import * as Deferred from "effect/Deferred" ;
1821import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter" ;
1922import * as Effect from "effect/Effect" ;
2023import * as Exit from "effect/Exit" ;
2124import * as FileSystem from "effect/FileSystem" ;
2225import * as Fiber from "effect/Fiber" ;
2326import * as Layer from "effect/Layer" ;
27+ import * as Queue from "effect/Queue" ;
2428import * as Ref from "effect/Ref" ;
2529import * as Stream from "effect/Stream" ;
2630import * as TestClock from "effect/testing/TestClock" ;
@@ -842,3 +846,270 @@ it.effect("serves the shared thread's files to its guest, never another thread's
842846 assert . isDefined ( shareId ) ;
843847 } ) . pipe ( Effect . scoped , Effect . provide ( layerTest ) ) ,
844848) ;
849+
850+ it . effect ( "serves an image on the host only when an agent reply the guest sees shows it" , ( ) =>
851+ Effect . gen ( function * ( ) {
852+ const { orchestrator, guestShares } = yield * shareThread ;
853+ const runId = yield * ownerRun ;
854+ const fileSystem = yield * FileSystem . FileSystem ;
855+ const directory = yield * fileSystem . makeTempDirectoryScoped ( ) ;
856+ const shot = `${ directory } /my shot.png` ;
857+ const fileUrl = `${ directory } /file-url.png` ;
858+ const page = `${ directory } /page.png` ;
859+ const fromUser = `${ directory } /from-user.png` ;
860+ const unnamed = `${ directory } /unnamed.png` ;
861+ const later = `${ directory } /later.png` ;
862+ for ( const path of [ shot , fileUrl , fromUser , unnamed , later ] ) {
863+ yield * fileSystem . writeFile ( path , PNG ) ;
864+ }
865+ yield * fileSystem . writeFileString ( page , "<script>alert(1)</script>" ) ;
866+ const eventSink = yield * EventSink . EventSinkV2 ;
867+ const now = yield * DateTime . now ;
868+ const run = ( yield * orchestrator . getThreadRecords ( threadId , [ "runs" ] ) ) . runs [ 0 ] ;
869+ const item = (
870+ id : string ,
871+ ordinal : number ,
872+ fields :
873+ | { readonly type : "assistant_message" ; readonly text : string }
874+ | { readonly type : "user_message" ; readonly text : string } ,
875+ target : { readonly threadId : ThreadId ; readonly runId : RunId } = { threadId, runId } ,
876+ ) => ( {
877+ id : EventId . make ( `event:${ id } ` ) ,
878+ type : "turn-item.updated" as const ,
879+ threadId : target . threadId ,
880+ driver : ProviderDriverKind . make ( "codex" ) ,
881+ providerInstanceId : instanceId ,
882+ occurredAt : now ,
883+ payload : {
884+ id : TurnItemId . make ( id ) ,
885+ threadId : target . threadId ,
886+ runId : target . runId ,
887+ nodeId : null ,
888+ providerThreadId : null ,
889+ providerTurnId : null ,
890+ nativeItemRef : null ,
891+ parentItemId : null ,
892+ ordinal,
893+ title : null ,
894+ status : "completed" as const ,
895+ startedAt : now ,
896+ completedAt : now ,
897+ updatedAt : now ,
898+ messageId : MessageId . make ( id ) ,
899+ ...( fields . type === "assistant_message"
900+ ? { ...fields , streaming : false }
901+ : {
902+ ...fields ,
903+ createdBy : "user" as const ,
904+ creationSource : "web" as const ,
905+ inputIntent : "turn_start" as const ,
906+ attachments : [ ] ,
907+ } ) ,
908+ } ,
909+ } ) ;
910+ yield * eventSink . write ( {
911+ events : [
912+ item ( "item:agent-images" , 1 , {
913+ type : "assistant_message" ,
914+ text : `Done:   } ) ` ,
915+ } ) ,
916+ // A guest's sent message is a user message; it must not open files on the host.
917+ item ( "item:guest-image" , 2 , { type : "user_message" , text : `` } ) ,
918+ {
919+ id : EventId . make ( "event:run-done" ) ,
920+ type : "run.updated" as const ,
921+ threadId,
922+ runId,
923+ providerInstanceId : instanceId ,
924+ occurredAt : now ,
925+ payload : { ...run ! , status : "completed" as const , completedAt : now } ,
926+ } ,
927+ ] ,
928+ } ) ;
929+ // The guest watches a fork, which shows these replies as inherited history.
930+ const forkId = ThreadId . make ( "thread:guest-share-fork" ) ;
931+ yield * orchestrator . dispatch ( {
932+ type : "thread.fork" ,
933+ commandId : CommandId . make ( "fork" ) ,
934+ sourceThreadId : threadId ,
935+ targetThreadId : forkId ,
936+ sourcePoint : { type : "run" , runId } ,
937+ createdBy : "user" ,
938+ creationSource : "web" ,
939+ } ) ;
940+ const { url } = yield * guestShares . create ( { threadId : forkId } ) ;
941+ const { token : session } = yield * guestShares . redeem ( {
942+ credential : url . slice ( url . indexOf ( "#" ) + 1 ) ,
943+ name : "Sam" ,
944+ } ) ;
945+
946+ // The guest's page opens, and its stream sends the replies the fork inherits.
947+ const sent = yield * Queue . unbounded < GuestStreamEvent > ( ) ;
948+ yield * Stream . runForEach ( yield * guestShares . stream ( session ) , ( event ) =>
949+ Queue . offer ( sent , event ) ,
950+ ) . pipe ( Effect . forkScoped ) ;
951+ const sentItem = Effect . fn ( function * ( id : string ) {
952+ for ( ; ; ) {
953+ const event = yield * Queue . take ( sent ) ;
954+ if (
955+ event . type === "thread" &&
956+ event . item . kind === "event" &&
957+ event . item . event . type === "turn-item.updated" &&
958+ event . item . event . payload . id === id
959+ ) {
960+ return ;
961+ }
962+ }
963+ } ) ;
964+
965+ // The agent shows another screenshot live, then enough rows to push the first reply out of
966+ // the window, before the page asks for any image.
967+ yield * orchestrator . dispatch ( {
968+ type : "message.dispatch" ,
969+ commandId : CommandId . make ( "fork-message" ) ,
970+ threadId : forkId ,
971+ messageId : MessageId . make ( "message:fork" ) ,
972+ text : "One more" ,
973+ attachments : [ ] ,
974+ dispatchMode : { type : "defer_start" } ,
975+ createdBy : "user" ,
976+ creationSource : "web" ,
977+ } ) ;
978+ const forkRun = ( yield * orchestrator . getThreadRecords ( forkId , [ "runs" ] ) ) . runs . at ( - 1 ) ! ;
979+ const inFork = { threadId : forkId , runId : forkRun . id } ;
980+ yield * eventSink . write ( {
981+ events : [
982+ item (
983+ "item:later-image" ,
984+ 1 ,
985+ { type : "assistant_message" , text : `` } ,
986+ inFork ,
987+ ) ,
988+ ...Array . from ( { length : 600 } , ( _ , index ) =>
989+ item (
990+ `item:filler-${ index } ` ,
991+ index + 2 ,
992+ { type : "assistant_message" , text : "Working." } ,
993+ inFork ,
994+ ) ,
995+ ) ,
996+ ] ,
997+ } ) ;
998+ yield * sentItem ( "item:later-image" ) ;
999+
1000+ // The page still shows every reply it was sent, so each image loads.
1001+ for ( const path of [ shot , fileUrl , later ] ) {
1002+ assert . deepStrictEqual ( yield * guestShares . media ( session , path ) , {
1003+ path,
1004+ mimeType : "image/png" ,
1005+ } ) ;
1006+ }
1007+ for ( const path of [ page , fromUser , unnamed , `${ directory } /missing.png` ] ) {
1008+ assert . equal ( ( yield * Effect . flip ( guestShares . media ( session , path ) ) ) . reason , "unavailable" ) ;
1009+ }
1010+
1011+ // A reply edited to drop its image stops allowing it.
1012+ yield * eventSink . write ( {
1013+ events : [
1014+ {
1015+ ...item ( "item:later-image" , 1 , { type : "assistant_message" , text : "Gone." } , inFork ) ,
1016+ id : EventId . make ( "event:later-image-edited" ) ,
1017+ } ,
1018+ ] ,
1019+ } ) ;
1020+ yield * sentItem ( "item:later-image" ) ;
1021+ assert . equal ( ( yield * Effect . flip ( guestShares . media ( session , later ) ) ) . reason , "unavailable" ) ;
1022+ } ) . pipe ( Effect . scoped , Effect . provide ( layerTest ) ) ,
1023+ ) ;
1024+
1025+ it . effect ( "keeps a connected guest's images while other shares come and go" , ( ) =>
1026+ Effect . gen ( function * ( ) {
1027+ const fileSystem = yield * FileSystem . FileSystem ;
1028+ const directory = yield * fileSystem . makeTempDirectoryScoped ( ) ;
1029+ const shot = `${ directory } /shot.png` ;
1030+ yield * fileSystem . writeFile ( shot , PNG ) ;
1031+ const eventSink = yield * EventSink . EventSinkV2 ;
1032+ const now = yield * DateTime . now ;
1033+ /** A shared thread whose agent reply shows the screenshot, and its guest's session. */
1034+ const sharedWithImage = Effect . fn ( function * ( id : ThreadId ) {
1035+ const { orchestrator, guestShares, credential } = yield * shareThreadFor ( id ) ;
1036+ yield * orchestrator . dispatch ( {
1037+ type : "message.dispatch" ,
1038+ commandId : CommandId . make ( `message:${ id } ` ) ,
1039+ threadId : id ,
1040+ messageId : MessageId . make ( `message:${ id } ` ) ,
1041+ text : "Show me" ,
1042+ attachments : [ ] ,
1043+ dispatchMode : { type : "defer_start" } ,
1044+ createdBy : "user" ,
1045+ creationSource : "web" ,
1046+ } ) ;
1047+ const runId = ( yield * orchestrator . getThreadRecords ( id , [ "runs" ] ) ) . runs [ 0 ] ! . id ;
1048+ yield * eventSink . write ( {
1049+ events : [
1050+ {
1051+ id : EventId . make ( `event:reply:${ id } ` ) ,
1052+ type : "turn-item.updated" ,
1053+ threadId : id ,
1054+ driver : ProviderDriverKind . make ( "codex" ) ,
1055+ providerInstanceId : instanceId ,
1056+ occurredAt : now ,
1057+ payload : {
1058+ id : TurnItemId . make ( `item:reply:${ id } ` ) ,
1059+ threadId : id ,
1060+ runId,
1061+ nodeId : null ,
1062+ providerThreadId : null ,
1063+ providerTurnId : null ,
1064+ nativeItemRef : null ,
1065+ parentItemId : null ,
1066+ ordinal : 1 ,
1067+ title : null ,
1068+ status : "completed" ,
1069+ startedAt : now ,
1070+ completedAt : now ,
1071+ updatedAt : now ,
1072+ type : "assistant_message" ,
1073+ messageId : MessageId . make ( `item:reply:${ id } ` ) ,
1074+ text : `` ,
1075+ streaming : false ,
1076+ } ,
1077+ } ,
1078+ ] ,
1079+ } ) ;
1080+ const { token } = yield * guestShares . redeem ( { credential, name : "Sam" } ) ;
1081+ return { guestShares, token } ;
1082+ } ) ;
1083+
1084+ // This guest's page stays open while more guests than the record keeps open and close theirs.
1085+ const { guestShares, token } = yield * sharedWithImage ( threadId ) ;
1086+ const opened = yield * Deferred . make < void > ( ) ;
1087+ yield * Stream . runForEach ( yield * guestShares . stream ( token ) , ( ) =>
1088+ Deferred . succeed ( opened , undefined ) ,
1089+ ) . pipe ( Effect . forkScoped ) ;
1090+ yield * Deferred . await ( opened ) ;
1091+ const closedTokens : Array < string > = [ ] ;
1092+ for ( let index = 0 ; index < 65 ; index ++ ) {
1093+ const other = yield * sharedWithImage ( ThreadId . make ( `thread:other-${ index } ` ) ) ;
1094+ const events = yield * guestShares . stream ( other . token ) ;
1095+ // The first is built but never read, as for a HEAD request; the rest open and close.
1096+ if ( index > 0 ) yield * Stream . runDrain ( Stream . take ( events , 2 ) ) ;
1097+ closedTokens . push ( other . token ) ;
1098+ }
1099+ assert . equal ( ( yield * guestShares . media ( token , shot ) ) . mimeType , "image/png" ) ;
1100+ // Records of guests with no open page were dropped to make room.
1101+ for ( const closed of closedTokens . slice ( 0 , 2 ) ) {
1102+ assert . equal ( ( yield * Effect . flip ( guestShares . media ( closed , shot ) ) ) . reason , "unavailable" ) ;
1103+ }
1104+
1105+ // A new link on the thread starts empty: its guest gets nothing the last guest was sent.
1106+ const { url } = yield * guestShares . create ( { threadId } ) ;
1107+ const next = yield * guestShares . redeem ( {
1108+ credential : url . slice ( url . indexOf ( "#" ) + 1 ) ,
1109+ name : "Kim" ,
1110+ } ) ;
1111+ assert . equal ( ( yield * Effect . flip ( guestShares . media ( next . token , shot ) ) ) . reason , "unavailable" ) ;
1112+ yield * Stream . runDrain ( Stream . take ( yield * guestShares . stream ( next . token ) , 2 ) ) ;
1113+ assert . equal ( ( yield * guestShares . media ( next . token , shot ) ) . mimeType , "image/png" ) ;
1114+ } ) . pipe ( Effect . scoped , Effect . provide ( layerTest ) ) ,
1115+ ) ;
0 commit comments