diff --git a/packages/prime-sandboxes/src/prime_sandboxes/_connectrpc.py b/packages/prime-sandboxes/src/prime_sandboxes/_connectrpc.py index 27ed3a69a..6e116797f 100644 --- a/packages/prime-sandboxes/src/prime_sandboxes/_connectrpc.py +++ b/packages/prime-sandboxes/src/prime_sandboxes/_connectrpc.py @@ -25,3 +25,20 @@ def _reject_legacy_connect_python(legacy_version: str | None) -> None: from connectrpc.compat import google_protobuf_binary_codec # noqa: E402 GOOGLE_PROTOBUF_BINARY_CODEC = google_protobuf_binary_codec() + +# Response-body read faults raised inside pyqwest's Rust core (string set +# verified against the installed pyqwest 0.8.0 extension binary). They reach +# fault classifiers inside ConnectError INTERNAL messages minted upstream +# (e.g. the gateway); a client-local pyqwest fault on a unary RPC surfaces as +# UNAVAILABLE, already retried by code. Classifiers match the markers +# case-insensitively as substrings to tell this transport-level class apart +# from a real server INTERNAL. "read cancelled" subsumes the longer sibling +# under substring matching; both stay listed as the verified source strings. +PYQWEST_BODY_READ_ERROR_MARKERS = frozenset( + { + "error reading content", + "error reading full content", + "response body read cancelled", + "read cancelled", + } +) diff --git a/packages/prime-sandboxes/src/prime_sandboxes/_proto/command_session/command_session_pb2.py b/packages/prime-sandboxes/src/prime_sandboxes/_proto/command_session/command_session_pb2.py index b029f2a79..ed5ec22ff 100644 --- a/packages/prime-sandboxes/src/prime_sandboxes/_proto/command_session/command_session_pb2.py +++ b/packages/prime-sandboxes/src/prime_sandboxes/_proto/command_session/command_session_pb2.py @@ -24,7 +24,7 @@ -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n%command_session/command_session.proto\x12\x0f\x63ommand_session\"R\n\x03PTY\x12\'\n\x04size\x18\x01 \x01(\x0b\x32\x19.command_session.PTY.Size\x1a\"\n\x04Size\x12\x0c\n\x04\x63ols\x18\x01 \x01(\r\x12\x0c\n\x04rows\x18\x02 \x01(\r\"\xa5\x01\n\x0b\x43ommandSpec\x12\x0b\n\x03\x63md\x18\x01 \x01(\t\x12\x0c\n\x04\x61rgs\x18\x02 \x03(\t\x12\x34\n\x04\x65nvs\x18\x03 \x03(\x0b\x32&.command_session.CommandSpec.EnvsEntry\x12\x10\n\x03\x63wd\x18\x04 \x01(\tH\x00\x88\x01\x01\x1a+\n\tEnvsEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x42\x06\n\x04_cwd\"\r\n\x0bListRequest\"j\n\x12\x43ommandSessionInfo\x12-\n\x07\x63ommand\x18\x01 \x01(\x0b\x32\x1c.command_session.CommandSpec\x12\x0b\n\x03pid\x18\x02 \x01(\r\x12\x10\n\x03tag\x18\x03 \x01(\tH\x00\x88\x01\x01\x42\x06\n\x04_tag\"E\n\x0cListResponse\x12\x35\n\x08sessions\x18\x01 \x03(\x0b\x32#.command_session.CommandSessionInfo\"\xa5\x01\n\x0cStartRequest\x12-\n\x07\x63ommand\x18\x01 \x01(\x0b\x32\x1c.command_session.CommandSpec\x12&\n\x03pty\x18\x02 \x01(\x0b\x32\x14.command_session.PTYH\x00\x88\x01\x01\x12\x10\n\x03tag\x18\x03 \x01(\tH\x01\x88\x01\x01\x12\x12\n\x05stdin\x18\x04 \x01(\x08H\x02\x88\x01\x01\x42\x06\n\x04_ptyB\x06\n\x04_tagB\x08\n\x06_stdin\"y\n\rUpdateRequest\x12\x38\n\x07session\x18\x01 \x01(\x0b\x32\'.command_session.CommandSessionSelector\x12&\n\x03pty\x18\x02 \x01(\x0b\x32\x14.command_session.PTYH\x00\x88\x01\x01\x42\x06\n\x04_pty\"\x10\n\x0eUpdateResponse\"\xf2\x03\n\x13\x43ommandSessionEvent\x12@\n\x05start\x18\x01 \x01(\x0b\x32/.command_session.CommandSessionEvent.StartEventH\x00\x12>\n\x04\x64\x61ta\x18\x02 \x01(\x0b\x32..command_session.CommandSessionEvent.DataEventH\x00\x12<\n\x03\x65nd\x18\x03 \x01(\x0b\x32-.command_session.CommandSessionEvent.EndEventH\x00\x12\x43\n\tkeepalive\x18\x04 \x01(\x0b\x32..command_session.CommandSessionEvent.KeepAliveH\x00\x1a\x19\n\nStartEvent\x12\x0b\n\x03pid\x18\x01 \x01(\r\x1aH\n\tDataEvent\x12\x10\n\x06stdout\x18\x01 \x01(\x0cH\x00\x12\x10\n\x06stderr\x18\x02 \x01(\x0cH\x00\x12\r\n\x03pty\x18\x03 \x01(\x0cH\x00\x42\x08\n\x06output\x1a[\n\x08\x45ndEvent\x12\x11\n\texit_code\x18\x01 \x01(\x11\x12\x0e\n\x06\x65xited\x18\x02 \x01(\x08\x12\x0e\n\x06status\x18\x03 \x01(\t\x12\x12\n\x05\x65rror\x18\x04 \x01(\tH\x00\x88\x01\x01\x42\x08\n\x06_error\x1a\x0b\n\tKeepAliveB\x07\n\x05\x65vent\"D\n\rStartResponse\x12\x33\n\x05\x65vent\x18\x01 \x01(\x0b\x32$.command_session.CommandSessionEvent\"F\n\x0f\x43onnectResponse\x12\x33\n\x05\x65vent\x18\x01 \x01(\x0b\x32$.command_session.CommandSessionEvent\"z\n\x10SendInputRequest\x12\x38\n\x07session\x18\x01 \x01(\x0b\x32\'.command_session.CommandSessionSelector\x12,\n\x05input\x18\x02 \x01(\x0b\x32\x1d.command_session.CommandInput\"\x13\n\x11SendInputResponse\"7\n\x0c\x43ommandInput\x12\x0f\n\x05stdin\x18\x01 \x01(\x0cH\x00\x12\r\n\x03pty\x18\x02 \x01(\x0cH\x00\x42\x07\n\x05input\"\xf1\x02\n\x12StreamInputRequest\x12?\n\x05start\x18\x01 \x01(\x0b\x32..command_session.StreamInputRequest.StartEventH\x00\x12=\n\x04\x64\x61ta\x18\x02 \x01(\x0b\x32-.command_session.StreamInputRequest.DataEventH\x00\x12\x42\n\tkeepalive\x18\x03 \x01(\x0b\x32-.command_session.StreamInputRequest.KeepAliveH\x00\x1a\x46\n\nStartEvent\x12\x38\n\x07session\x18\x01 \x01(\x0b\x32\'.command_session.CommandSessionSelector\x1a\x39\n\tDataEvent\x12,\n\x05input\x18\x02 \x01(\x0b\x32\x1d.command_session.CommandInput\x1a\x0b\n\tKeepAliveB\x07\n\x05\x65vent\"\x15\n\x13StreamInputResponse\"v\n\x11SendSignalRequest\x12\x38\n\x07session\x18\x01 \x01(\x0b\x32\'.command_session.CommandSessionSelector\x12\'\n\x06signal\x18\x02 \x01(\x0e\x32\x17.command_session.Signal\"\x14\n\x12SendSignalResponse\"J\n\x0e\x43onnectRequest\x12\x38\n\x07session\x18\x01 \x01(\x0b\x32\'.command_session.CommandSessionSelector\"B\n\x16\x43ommandSessionSelector\x12\r\n\x03pid\x18\x01 \x01(\rH\x00\x12\r\n\x03tag\x18\x02 \x01(\tH\x00\x42\n\n\x08selector*H\n\x06Signal\x12\x16\n\x12SIGNAL_UNSPECIFIED\x10\x00\x12\x12\n\x0eSIGNAL_SIGTERM\x10\x0f\x12\x12\n\x0eSIGNAL_SIGKILL\x10\t2\xc1\x04\n\x0e\x43ommandSession\x12\x43\n\x04List\x12\x1c.command_session.ListRequest\x1a\x1d.command_session.ListResponse\x12N\n\x07\x43onnect\x12\x1f.command_session.ConnectRequest\x1a .command_session.ConnectResponse0\x01\x12H\n\x05Start\x12\x1d.command_session.StartRequest\x1a\x1e.command_session.StartResponse0\x01\x12I\n\x06Update\x12\x1e.command_session.UpdateRequest\x1a\x1f.command_session.UpdateResponse\x12Z\n\x0bStreamInput\x12#.command_session.StreamInputRequest\x1a$.command_session.StreamInputResponse(\x01\x12R\n\tSendInput\x12!.command_session.SendInputRequest\x1a\".command_session.SendInputResponse\x12U\n\nSendSignal\x12\".command_session.SendSignalRequest\x1a#.command_session.SendSignalResponseb\x06proto3') +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n%command_session/command_session.proto\x12\x0f\x63ommand_session\"R\n\x03PTY\x12\'\n\x04size\x18\x01 \x01(\x0b\x32\x19.command_session.PTY.Size\x1a\"\n\x04Size\x12\x0c\n\x04\x63ols\x18\x01 \x01(\r\x12\x0c\n\x04rows\x18\x02 \x01(\r\"\xa5\x01\n\x0b\x43ommandSpec\x12\x0b\n\x03\x63md\x18\x01 \x01(\t\x12\x0c\n\x04\x61rgs\x18\x02 \x03(\t\x12\x34\n\x04\x65nvs\x18\x03 \x03(\x0b\x32&.command_session.CommandSpec.EnvsEntry\x12\x10\n\x03\x63wd\x18\x04 \x01(\tH\x00\x88\x01\x01\x1a+\n\tEnvsEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x42\x06\n\x04_cwd\"\r\n\x0bListRequest\"\x87\x01\n\x12\x43ommandSessionInfo\x12-\n\x07\x63ommand\x18\x01 \x01(\x0b\x32\x1c.command_session.CommandSpec\x12\x0b\n\x03pid\x18\x02 \x01(\r\x12\x19\n\x0csession_uuid\x18\x04 \x01(\tH\x00\x88\x01\x01\x42\x0f\n\r_session_uuidJ\x04\x08\x03\x10\x04R\x03tag\"E\n\x0cListResponse\x12\x35\n\x08sessions\x18\x01 \x03(\x0b\x32#.command_session.CommandSessionInfo\"\xc2\x01\n\x0cStartRequest\x12-\n\x07\x63ommand\x18\x01 \x01(\x0b\x32\x1c.command_session.CommandSpec\x12&\n\x03pty\x18\x02 \x01(\x0b\x32\x14.command_session.PTYH\x00\x88\x01\x01\x12\x12\n\x05stdin\x18\x04 \x01(\x08H\x01\x88\x01\x01\x12\x19\n\x0csession_uuid\x18\x05 \x01(\tH\x02\x88\x01\x01\x42\x06\n\x04_ptyB\x08\n\x06_stdinB\x0f\n\r_session_uuidJ\x04\x08\x03\x10\x04R\x03tag\"y\n\rUpdateRequest\x12\x38\n\x07session\x18\x01 \x01(\x0b\x32\'.command_session.CommandSessionSelector\x12&\n\x03pty\x18\x02 \x01(\x0b\x32\x14.command_session.PTYH\x00\x88\x01\x01\x42\x06\n\x04_pty\"\x10\n\x0eUpdateResponse\"\xf2\x03\n\x13\x43ommandSessionEvent\x12@\n\x05start\x18\x01 \x01(\x0b\x32/.command_session.CommandSessionEvent.StartEventH\x00\x12>\n\x04\x64\x61ta\x18\x02 \x01(\x0b\x32..command_session.CommandSessionEvent.DataEventH\x00\x12<\n\x03\x65nd\x18\x03 \x01(\x0b\x32-.command_session.CommandSessionEvent.EndEventH\x00\x12\x43\n\tkeepalive\x18\x04 \x01(\x0b\x32..command_session.CommandSessionEvent.KeepAliveH\x00\x1a\x19\n\nStartEvent\x12\x0b\n\x03pid\x18\x01 \x01(\r\x1aH\n\tDataEvent\x12\x10\n\x06stdout\x18\x01 \x01(\x0cH\x00\x12\x10\n\x06stderr\x18\x02 \x01(\x0cH\x00\x12\r\n\x03pty\x18\x03 \x01(\x0cH\x00\x42\x08\n\x06output\x1a[\n\x08\x45ndEvent\x12\x11\n\texit_code\x18\x01 \x01(\x11\x12\x0e\n\x06\x65xited\x18\x02 \x01(\x08\x12\x0e\n\x06status\x18\x03 \x01(\t\x12\x12\n\x05\x65rror\x18\x04 \x01(\tH\x00\x88\x01\x01\x42\x08\n\x06_error\x1a\x0b\n\tKeepAliveB\x07\n\x05\x65vent\"D\n\rStartResponse\x12\x33\n\x05\x65vent\x18\x01 \x01(\x0b\x32$.command_session.CommandSessionEvent\"F\n\x0f\x43onnectResponse\x12\x33\n\x05\x65vent\x18\x01 \x01(\x0b\x32$.command_session.CommandSessionEvent\"\xa2\x01\n\x10SendInputRequest\x12\x38\n\x07session\x18\x01 \x01(\x0b\x32\'.command_session.CommandSessionSelector\x12,\n\x05input\x18\x02 \x01(\x0b\x32\x1d.command_session.CommandInput\x12\x17\n\ninput_uuid\x18\x03 \x01(\tH\x00\x88\x01\x01\x42\r\n\x0b_input_uuid\"\x13\n\x11SendInputResponse\"7\n\x0c\x43ommandInput\x12\x0f\n\x05stdin\x18\x01 \x01(\x0cH\x00\x12\r\n\x03pty\x18\x02 \x01(\x0cH\x00\x42\x07\n\x05input\"\xf1\x02\n\x12StreamInputRequest\x12?\n\x05start\x18\x01 \x01(\x0b\x32..command_session.StreamInputRequest.StartEventH\x00\x12=\n\x04\x64\x61ta\x18\x02 \x01(\x0b\x32-.command_session.StreamInputRequest.DataEventH\x00\x12\x42\n\tkeepalive\x18\x03 \x01(\x0b\x32-.command_session.StreamInputRequest.KeepAliveH\x00\x1a\x46\n\nStartEvent\x12\x38\n\x07session\x18\x01 \x01(\x0b\x32\'.command_session.CommandSessionSelector\x1a\x39\n\tDataEvent\x12,\n\x05input\x18\x01 \x01(\x0b\x32\x1d.command_session.CommandInput\x1a\x0b\n\tKeepAliveB\x07\n\x05\x65vent\"\x15\n\x13StreamInputResponse\"\xa0\x01\n\x11SendSignalRequest\x12\x38\n\x07session\x18\x01 \x01(\x0b\x32\'.command_session.CommandSessionSelector\x12\'\n\x06signal\x18\x02 \x01(\x0e\x32\x17.command_session.Signal\x12\x18\n\x0bsignal_uuid\x18\x03 \x01(\tH\x00\x88\x01\x01\x42\x0e\n\x0c_signal_uuid\"\x14\n\x12SendSignalResponse\"J\n\x0e\x43onnectRequest\x12\x38\n\x07session\x18\x01 \x01(\x0b\x32\'.command_session.CommandSessionSelector\"V\n\x16\x43ommandSessionSelector\x12\r\n\x03pid\x18\x01 \x01(\rH\x00\x12\x16\n\x0csession_uuid\x18\x03 \x01(\tH\x00\x42\n\n\x08selectorJ\x04\x08\x02\x10\x03R\x03tag*H\n\x06Signal\x12\x16\n\x12SIGNAL_UNSPECIFIED\x10\x00\x12\x12\n\x0eSIGNAL_SIGTERM\x10\x0f\x12\x12\n\x0eSIGNAL_SIGKILL\x10\t2\xc1\x04\n\x0e\x43ommandSession\x12\x43\n\x04List\x12\x1c.command_session.ListRequest\x1a\x1d.command_session.ListResponse\x12N\n\x07\x43onnect\x12\x1f.command_session.ConnectRequest\x1a .command_session.ConnectResponse0\x01\x12H\n\x05Start\x12\x1d.command_session.StartRequest\x1a\x1e.command_session.StartResponse0\x01\x12I\n\x06Update\x12\x1e.command_session.UpdateRequest\x1a\x1f.command_session.UpdateResponse\x12Z\n\x0bStreamInput\x12#.command_session.StreamInputRequest\x1a$.command_session.StreamInputResponse(\x01\x12R\n\tSendInput\x12!.command_session.SendInputRequest\x1a\".command_session.SendInputResponse\x12U\n\nSendSignal\x12\".command_session.SendSignalRequest\x1a#.command_session.SendSignalResponseb\x06proto3') _globals = globals() _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) @@ -33,8 +33,8 @@ DESCRIPTOR._loaded_options = None _globals['_COMMANDSPEC_ENVSENTRY']._loaded_options = None _globals['_COMMANDSPEC_ENVSENTRY']._serialized_options = b'8\001' - _globals['_SIGNAL']._serialized_start=2339 - _globals['_SIGNAL']._serialized_end=2411 + _globals['_SIGNAL']._serialized_start=2502 + _globals['_SIGNAL']._serialized_end=2574 _globals['_PTY']._serialized_start=58 _globals['_PTY']._serialized_end=140 _globals['_PTY_SIZE']._serialized_start=106 @@ -45,54 +45,54 @@ _globals['_COMMANDSPEC_ENVSENTRY']._serialized_end=300 _globals['_LISTREQUEST']._serialized_start=310 _globals['_LISTREQUEST']._serialized_end=323 - _globals['_COMMANDSESSIONINFO']._serialized_start=325 - _globals['_COMMANDSESSIONINFO']._serialized_end=431 - _globals['_LISTRESPONSE']._serialized_start=433 - _globals['_LISTRESPONSE']._serialized_end=502 - _globals['_STARTREQUEST']._serialized_start=505 - _globals['_STARTREQUEST']._serialized_end=670 - _globals['_UPDATEREQUEST']._serialized_start=672 - _globals['_UPDATEREQUEST']._serialized_end=793 - _globals['_UPDATERESPONSE']._serialized_start=795 - _globals['_UPDATERESPONSE']._serialized_end=811 - _globals['_COMMANDSESSIONEVENT']._serialized_start=814 - _globals['_COMMANDSESSIONEVENT']._serialized_end=1312 - _globals['_COMMANDSESSIONEVENT_STARTEVENT']._serialized_start=1098 - _globals['_COMMANDSESSIONEVENT_STARTEVENT']._serialized_end=1123 - _globals['_COMMANDSESSIONEVENT_DATAEVENT']._serialized_start=1125 - _globals['_COMMANDSESSIONEVENT_DATAEVENT']._serialized_end=1197 - _globals['_COMMANDSESSIONEVENT_ENDEVENT']._serialized_start=1199 - _globals['_COMMANDSESSIONEVENT_ENDEVENT']._serialized_end=1290 - _globals['_COMMANDSESSIONEVENT_KEEPALIVE']._serialized_start=1292 - _globals['_COMMANDSESSIONEVENT_KEEPALIVE']._serialized_end=1303 - _globals['_STARTRESPONSE']._serialized_start=1314 - _globals['_STARTRESPONSE']._serialized_end=1382 - _globals['_CONNECTRESPONSE']._serialized_start=1384 - _globals['_CONNECTRESPONSE']._serialized_end=1454 - _globals['_SENDINPUTREQUEST']._serialized_start=1456 - _globals['_SENDINPUTREQUEST']._serialized_end=1578 - _globals['_SENDINPUTRESPONSE']._serialized_start=1580 - _globals['_SENDINPUTRESPONSE']._serialized_end=1599 - _globals['_COMMANDINPUT']._serialized_start=1601 - _globals['_COMMANDINPUT']._serialized_end=1656 - _globals['_STREAMINPUTREQUEST']._serialized_start=1659 - _globals['_STREAMINPUTREQUEST']._serialized_end=2028 - _globals['_STREAMINPUTREQUEST_STARTEVENT']._serialized_start=1877 - _globals['_STREAMINPUTREQUEST_STARTEVENT']._serialized_end=1947 - _globals['_STREAMINPUTREQUEST_DATAEVENT']._serialized_start=1949 - _globals['_STREAMINPUTREQUEST_DATAEVENT']._serialized_end=2006 - _globals['_STREAMINPUTREQUEST_KEEPALIVE']._serialized_start=1292 - _globals['_STREAMINPUTREQUEST_KEEPALIVE']._serialized_end=1303 - _globals['_STREAMINPUTRESPONSE']._serialized_start=2030 - _globals['_STREAMINPUTRESPONSE']._serialized_end=2051 - _globals['_SENDSIGNALREQUEST']._serialized_start=2053 - _globals['_SENDSIGNALREQUEST']._serialized_end=2171 - _globals['_SENDSIGNALRESPONSE']._serialized_start=2173 - _globals['_SENDSIGNALRESPONSE']._serialized_end=2193 - _globals['_CONNECTREQUEST']._serialized_start=2195 - _globals['_CONNECTREQUEST']._serialized_end=2269 - _globals['_COMMANDSESSIONSELECTOR']._serialized_start=2271 - _globals['_COMMANDSESSIONSELECTOR']._serialized_end=2337 - _globals['_COMMANDSESSION']._serialized_start=2414 - _globals['_COMMANDSESSION']._serialized_end=2991 + _globals['_COMMANDSESSIONINFO']._serialized_start=326 + _globals['_COMMANDSESSIONINFO']._serialized_end=461 + _globals['_LISTRESPONSE']._serialized_start=463 + _globals['_LISTRESPONSE']._serialized_end=532 + _globals['_STARTREQUEST']._serialized_start=535 + _globals['_STARTREQUEST']._serialized_end=729 + _globals['_UPDATEREQUEST']._serialized_start=731 + _globals['_UPDATEREQUEST']._serialized_end=852 + _globals['_UPDATERESPONSE']._serialized_start=854 + _globals['_UPDATERESPONSE']._serialized_end=870 + _globals['_COMMANDSESSIONEVENT']._serialized_start=873 + _globals['_COMMANDSESSIONEVENT']._serialized_end=1371 + _globals['_COMMANDSESSIONEVENT_STARTEVENT']._serialized_start=1157 + _globals['_COMMANDSESSIONEVENT_STARTEVENT']._serialized_end=1182 + _globals['_COMMANDSESSIONEVENT_DATAEVENT']._serialized_start=1184 + _globals['_COMMANDSESSIONEVENT_DATAEVENT']._serialized_end=1256 + _globals['_COMMANDSESSIONEVENT_ENDEVENT']._serialized_start=1258 + _globals['_COMMANDSESSIONEVENT_ENDEVENT']._serialized_end=1349 + _globals['_COMMANDSESSIONEVENT_KEEPALIVE']._serialized_start=1351 + _globals['_COMMANDSESSIONEVENT_KEEPALIVE']._serialized_end=1362 + _globals['_STARTRESPONSE']._serialized_start=1373 + _globals['_STARTRESPONSE']._serialized_end=1441 + _globals['_CONNECTRESPONSE']._serialized_start=1443 + _globals['_CONNECTRESPONSE']._serialized_end=1513 + _globals['_SENDINPUTREQUEST']._serialized_start=1516 + _globals['_SENDINPUTREQUEST']._serialized_end=1678 + _globals['_SENDINPUTRESPONSE']._serialized_start=1680 + _globals['_SENDINPUTRESPONSE']._serialized_end=1699 + _globals['_COMMANDINPUT']._serialized_start=1701 + _globals['_COMMANDINPUT']._serialized_end=1756 + _globals['_STREAMINPUTREQUEST']._serialized_start=1759 + _globals['_STREAMINPUTREQUEST']._serialized_end=2128 + _globals['_STREAMINPUTREQUEST_STARTEVENT']._serialized_start=1977 + _globals['_STREAMINPUTREQUEST_STARTEVENT']._serialized_end=2047 + _globals['_STREAMINPUTREQUEST_DATAEVENT']._serialized_start=2049 + _globals['_STREAMINPUTREQUEST_DATAEVENT']._serialized_end=2106 + _globals['_STREAMINPUTREQUEST_KEEPALIVE']._serialized_start=1351 + _globals['_STREAMINPUTREQUEST_KEEPALIVE']._serialized_end=1362 + _globals['_STREAMINPUTRESPONSE']._serialized_start=2130 + _globals['_STREAMINPUTRESPONSE']._serialized_end=2151 + _globals['_SENDSIGNALREQUEST']._serialized_start=2154 + _globals['_SENDSIGNALREQUEST']._serialized_end=2314 + _globals['_SENDSIGNALRESPONSE']._serialized_start=2316 + _globals['_SENDSIGNALRESPONSE']._serialized_end=2336 + _globals['_CONNECTREQUEST']._serialized_start=2338 + _globals['_CONNECTREQUEST']._serialized_end=2412 + _globals['_COMMANDSESSIONSELECTOR']._serialized_start=2414 + _globals['_COMMANDSESSIONSELECTOR']._serialized_end=2500 + _globals['_COMMANDSESSION']._serialized_start=2577 + _globals['_COMMANDSESSION']._serialized_end=3154 # @@protoc_insertion_point(module_scope) diff --git a/packages/prime-sandboxes/src/prime_sandboxes/process.py b/packages/prime-sandboxes/src/prime_sandboxes/process.py index 78357e883..e1c6ca336 100644 --- a/packages/prime-sandboxes/src/prime_sandboxes/process.py +++ b/packages/prime-sandboxes/src/prime_sandboxes/process.py @@ -7,13 +7,12 @@ from typing import Literal from connectrpc.client import ConnectClient -from connectrpc.code import Code from connectrpc.errors import ConnectError from google.protobuf.message import Message from pyqwest import HTTPTransport from .core import APIError -from .rpc_command_session import parse_command_session_start_event +from .rpc_command_session import is_recoverable_stream_fault, parse_command_session_start_event logger = logging.getLogger(__name__) @@ -22,9 +21,11 @@ _STREAM_MAX_RECONNECTS = 5 _STREAM_RECONNECT_BACKOFF_SECONDS = 0.5 -_WriteStdin = Callable[[int, bytes], Awaitable[None]] -_SendSignal = Callable[[int, Literal["terminate", "kill"]], Awaitable[None]] -_Reconnect = Callable[[int | None], AsyncIterator[Message]] +_WriteStdin = Callable[[bytes], Awaitable[None]] +_SendSignal = Callable[[Literal["terminate", "kill"]], Awaitable[None]] +# The argument is whether a StartEvent has been observed yet; the callee picks +# between retrying Start (create-or-attach) and Connect-ing to the session. +_Reconnect = Callable[[bool], AsyncIterator[Message]] class _AsyncProcessStream(AsyncIterator[bytes]): @@ -148,7 +149,7 @@ async def write_stdin(self, data: bytes) -> None: return if self._closed or self._remote_exited: raise BrokenPipeError("process has exited") - await self._write_stdin(self.pid, data) + await self._write_stdin(data) async def wait(self) -> int: """Wait for the process to exit and return its exit code.""" @@ -165,7 +166,7 @@ async def kill(self) -> None: async def _send_signal(self, signal: Literal["terminate", "kill"]) -> None: if self._closed or self._remote_exited: return - await self._send_process_signal(self.pid, signal) + await self._send_process_signal(signal) self._signals_sent.add(signal) async def aclose(self) -> None: @@ -234,12 +235,10 @@ async def _wait_for_exit_event(self) -> bool: return self._remote_exited def _can_reconnect(self, reconnects: int, error: BaseException | None) -> bool: - """Whether the process has enough identity and budget for another attach.""" + """Whether the stream fault is recoverable and reconnect budget remains.""" if self._reconnect is None or self._remote_exited or reconnects >= _STREAM_MAX_RECONNECTS: return False - if not isinstance(error, ConnectError) or error.code != Code.NOT_FOUND: - return True - return not self._started.done() and reconnects > 0 + return is_recoverable_stream_fault(error) async def _aclose_stream(self) -> None: close = getattr(self._stream, "aclose", None) @@ -261,10 +260,11 @@ async def _reconnect_stream(self, reconnects: int, error: BaseException | None) ) await self._aclose_stream() await asyncio.sleep(delay) - # Connect tails the same process from re-attachment time; output emitted while detached - # is not replayed. - pid = self.pid if self._started.done() else None - self._stream = reconnect(pid) + # Re-attachment — a retried Start before the pid was seen, Connect after — + # never replays output emitted while detached. Both arms re-announce the + # StartEvent and replay the retained EndEvent of a session that exited + # within sandboxd's retention window, so a missed exit is still observed. + self._stream = reconnect(self._started.done()) async def _pump(self) -> None: ended = False @@ -279,7 +279,7 @@ async def _pump(self) -> None: continue kind, value = event if kind == "start": - # A reconnected (Connect) stream re-announces the pid; keep the first. + # A re-attached stream re-announces the pid; keep the first. if not self._started.done(): self._started.set_result(value) elif kind == "stdout": diff --git a/packages/prime-sandboxes/src/prime_sandboxes/rpc_command_session.py b/packages/prime-sandboxes/src/prime_sandboxes/rpc_command_session.py index 1d23b412c..6f4279b4c 100644 --- a/packages/prime-sandboxes/src/prime_sandboxes/rpc_command_session.py +++ b/packages/prime-sandboxes/src/prime_sandboxes/rpc_command_session.py @@ -2,9 +2,12 @@ from typing import Dict, List, Literal, Optional, Protocol, Sequence, cast +from connectrpc.code import Code +from connectrpc.errors import ConnectError from connectrpc.method import IdempotencyLevel, MethodInfo from google.protobuf.message import Message +from ._connectrpc import PYQWEST_BODY_READ_ERROR_MARKERS from ._proto.command_session import command_session_pb2 @@ -18,12 +21,12 @@ def __call__(self, *, cmd: str, args: List[str], envs: Dict[str, str]) -> _Comma class _CommandSessionStartRequestFactory(Protocol): def __call__( - self, *, command: _CommandSpecLike, stdin: bool, tag: str | None = None + self, *, command: _CommandSpecLike, stdin: bool, session_uuid: str | None = None ) -> Message: ... class _CommandSessionSelectorFactory(Protocol): - def __call__(self, *, pid: int) -> Message: ... + def __call__(self, *, session_uuid: str) -> Message: ... class _CommandInputFactory(Protocol): @@ -31,11 +34,11 @@ def __call__(self, *, stdin: bytes) -> Message: ... class _CommandSessionSendInputRequestFactory(Protocol): - def __call__(self, *, session: Message, input: Message) -> Message: ... + def __call__(self, *, session: Message, input: Message, input_uuid: str) -> Message: ... class _CommandSessionSendSignalRequestFactory(Protocol): - def __call__(self, *, session: Message, signal: int) -> Message: ... + def __call__(self, *, session: Message, signal: int, signal_uuid: str) -> Message: ... class _CommandSessionConnectRequestFactory(Protocol): @@ -44,7 +47,8 @@ def __call__(self, *, session: Message) -> Message: ... class _CommandSessionInfoLike(Protocol): pid: int - tag: str + session_uuid: str + command: _CommandSpecLike class _CommandSessionListResponseLike(Protocol): @@ -164,6 +168,8 @@ def HasField(self, field_name: str) -> bool: ... idempotency_level=IdempotencyLevel.NO_SIDE_EFFECTS, ) +# Live-process introspection: pid, session_uuid, and command for each running +# process. Permanent public API; exited sessions are not listed. COMMAND_SESSION_LIST_RPC_METHOD = MethodInfo( name="List", service_name="command_session.CommandSession", @@ -173,13 +179,48 @@ def HasField(self, field_name: str) -> bool: ... ) +# The two fault predicates below classify command-session RPC failures for +# retry, with deliberately opposite polarity. Stream re-attach (Connect, or a +# create-or-attach Start resending the identical request) is idempotent, so +# is_recoverable_stream_fault is a deny-list: retry everything except the codes +# command_session.proto promises as definitive answers. A unary control RPC's +# unknown fault may itself be a definitive answer, so is_transient_control_fault +# is an allow-list: fail fast on everything except known link faults. + +# Stream faults recovery cannot fix, per command_session.proto's code promises: +# NOT_FOUND (the session is gone or its retention expired) and +# FAILED_PRECONDITION (a Start reusing the session_uuid with a different spec — +# a guard for a future non-identical retry; today's reconnect resends the +# identical request, so the server cannot answer it with a spec conflict). +_STREAM_FATAL_CODES = frozenset({Code.NOT_FOUND, Code.FAILED_PRECONDITION}) + +# Link faults a unary control RPC may retry; pyqwest body-read faults surface +# as ConnectError INTERNAL and are matched by message marker instead. +_TRANSIENT_CONTROL_CODES = frozenset({Code.DEADLINE_EXCEEDED, Code.UNAVAILABLE}) + + +def is_recoverable_stream_fault(error: BaseException | None) -> bool: + """Whether a dropped command-session stream may be re-attached (None: clean EOF).""" + return not (isinstance(error, ConnectError) and error.code in _STREAM_FATAL_CODES) + + +def is_transient_control_fault(error: ConnectError) -> bool: + """Whether a unary control-RPC fault is a link hiccup rather than a definitive answer.""" + if error.code in _TRANSIENT_CONTROL_CODES: + return True + message = (error.message or "").lower() + return error.code == Code.INTERNAL and any( + marker in message for marker in PYQWEST_BODY_READ_ERROR_MARKERS + ) + + def build_command_session_start_request( + *, command: str, working_dir: Optional[str], env: Optional[Dict[str, str]], - *, stdin: bool = False, - tag: str | None = None, + session_uuid: str | None = None, ) -> Message: command_spec = _COMMAND_SPEC_FACTORY( cmd="/bin/bash", @@ -192,13 +233,7 @@ def build_command_session_start_request( return _COMMAND_SESSION_START_REQUEST_FACTORY( command=command_spec, stdin=stdin, - tag=tag, - ) - - -def build_command_session_connect_request(pid: int) -> Message: - return _COMMAND_SESSION_CONNECT_REQUEST_FACTORY( - session=_COMMAND_SESSION_SELECTOR_FACTORY(pid=pid) + session_uuid=session_uuid, ) @@ -206,28 +241,33 @@ def build_command_session_list_request() -> Message: return _COMMAND_SESSION_LIST_REQUEST_TYPE() -def find_command_session_pid(response: Message, tag: str) -> int | None: - sessions = cast(_CommandSessionListResponseLike, response).sessions - return next((int(session.pid) for session in sessions if session.tag == tag), None) +def build_command_session_connect_request(*, session_uuid: str) -> Message: + return _COMMAND_SESSION_CONNECT_REQUEST_FACTORY( + session=_COMMAND_SESSION_SELECTOR_FACTORY(session_uuid=session_uuid) + ) -def build_command_session_send_input_request(pid: int, data: bytes) -> Message: +def build_command_session_send_input_request( + *, session_uuid: str, data: bytes, input_uuid: str +) -> Message: return _COMMAND_SESSION_SEND_INPUT_REQUEST_FACTORY( - session=_COMMAND_SESSION_SELECTOR_FACTORY(pid=pid), + session=_COMMAND_SESSION_SELECTOR_FACTORY(session_uuid=session_uuid), input=_COMMAND_INPUT_FACTORY(stdin=data), + input_uuid=input_uuid, ) def build_command_session_send_signal_request( - pid: int, signal: Literal["terminate", "kill"] + *, session_uuid: str, signal: Literal["terminate", "kill"], signal_uuid: str ) -> Message: signal_value = getattr( command_session_pb2, "SIGNAL_SIGTERM" if signal == "terminate" else "SIGNAL_SIGKILL", ) return _COMMAND_SESSION_SEND_SIGNAL_REQUEST_FACTORY( - session=_COMMAND_SESSION_SELECTOR_FACTORY(pid=pid), + session=_COMMAND_SESSION_SELECTOR_FACTORY(session_uuid=session_uuid), signal=signal_value, + signal_uuid=signal_uuid, ) diff --git a/packages/prime-sandboxes/src/prime_sandboxes/sandbox.py b/packages/prime-sandboxes/src/prime_sandboxes/sandbox.py index 9f29ef49d..34840d612 100644 --- a/packages/prime-sandboxes/src/prime_sandboxes/sandbox.py +++ b/packages/prime-sandboxes/src/prime_sandboxes/sandbox.py @@ -88,17 +88,15 @@ from .process import AsyncSandboxProcess from .rpc_command_session import ( COMMAND_SESSION_CONNECT_RPC_METHOD, - COMMAND_SESSION_LIST_RPC_METHOD, COMMAND_SESSION_SEND_INPUT_RPC_METHOD, COMMAND_SESSION_SEND_SIGNAL_RPC_METHOD, COMMAND_SESSION_START_RPC_METHOD, build_command_session_connect_request, - build_command_session_list_request, build_command_session_send_input_request, build_command_session_send_signal_request, build_command_session_start_request, collect_command_session_start_event, - find_command_session_pid, + is_transient_control_fault, ) # Connection-level errors: request never reached the server, so retry is safe @@ -118,9 +116,15 @@ _LIVE_PROCESS_TIMEOUT_MS = 24 * 60 * 60 * 1000 _PROCESS_INPUT_TIMEOUT_MS = 30_000 _PROCESS_SIGNAL_TIMEOUT_MS = 10_000 -_PROCESS_DISCOVERY_TIMEOUT_MS = 10_000 _LIVE_PROCESS_TCP_KEEPALIVE_SECONDS = 15.0 _LIVE_PROCESS_POOL_IDLE_TIMEOUT_SECONDS = 300.0 + +# Live-process control RPC retry budget (see _execute_process_control_rpc). +# Worst-case retry horizon: 3 attempts x the 30s stdin RPC timeout plus +# 0.5s + 1s backoff ~= 91.5s. sandboxd's idempotency window must exceed it +# (invariant 4 in the platform's sandboxd-idempotency context doc). +_PROCESS_CONTROL_RPC_ATTEMPTS = 3 +_PROCESS_CONTROL_RETRY_INITIAL_DELAY = 0.5 _BACKGROUND_JOB_LAUNCH_ATTEMPTS = 3 _BACKGROUND_JOB_LAUNCH_BACKOFF_SECONDS = 0.5 _BACKGROUND_JOB_LAUNCH_TIMEOUT_SECONDS = 30 @@ -144,6 +148,11 @@ def _ca_bundle() -> bytes: return ca_file.read() +def _canonical_uuid_key() -> str: + """Mint one idempotency key; sandboxd rejects non-canonical UUID spellings.""" + return str(uuid.uuid4()) + + def _live_process_transport() -> HTTPTransport: # A bare HTTPTransport carries no trust roots on some pyqwest versions # (only the default singleton does), so pass certifi's bundle explicitly. @@ -1242,7 +1251,9 @@ def _execute_command_connect_rpc( timeout: Optional[int] = None, ) -> CommandResponse: effective_timeout = timeout if timeout is not None else 300 - request = build_command_session_start_request(command, working_dir, env) + request = build_command_session_start_request( + command=command, working_dir=working_dir, env=env + ) reauthed = False while True: @@ -2670,9 +2681,13 @@ async def open_process( The returned handle streams stdout and stderr, accepts stdin writes, waits for the exit code, and can signal the process. Container sandboxes - do not expose this transport and fail fast. If the initial stream drops - before reporting a PID, recovery can find the session only while it is - still running because sandboxd does not retain completed sessions. + do not expose this transport and fail fast. If the stream drops before + the first StartEvent, the SDK retries Start with the same session_uuid + (create-or-attach); after it, the SDK re-attaches with Connect by + session selector. Either way sandboxd re-announces the StartEvent and, + for a session that exited within its retention window, replays the + retained EndEvent — an exit missed while detached is still observed. + Output emitted while detached is not replayed. """ await self._auth_cache.get_or_refresh(sandbox_id) if not await self._auth_cache.is_vm(sandbox_id): @@ -2698,15 +2713,16 @@ async def open_process( send_compression=None, http_client=http_client, ) - # The SDK tag lets reconnect find a Start whose PID response was lost. Remove tag/List - # recovery once sandboxd provides idempotent Start/create-or-attach semantics. - process_tag = f"prime-sdk-{uuid.uuid4().hex}" + # session_uuid is the Start idempotency key: re-issuing the same request + # attaches to (or replays) the session instead of spawning a second + # process. + session_uuid = _canonical_uuid_key() request = build_command_session_start_request( - command, - working_dir, - env, + command=command, + working_dir=working_dir, + env=env, stdin=True, - tag=process_tag, + session_uuid=session_uuid, ) stream = rpc_client.execute_server_stream( request=request, @@ -2715,27 +2731,43 @@ async def open_process( timeout_ms=_LIVE_PROCESS_TIMEOUT_MS, ) - async def write_stdin(pid: int, data: bytes) -> None: + async def write_stdin(data: bytes) -> None: + input_uuid = _canonical_uuid_key() await self._execute_process_control_rpc( sandbox_id, - build_command_session_send_input_request(pid, data), + build_command_session_send_input_request( + session_uuid=session_uuid, data=data, input_uuid=input_uuid + ), COMMAND_SESSION_SEND_INPUT_RPC_METHOD, _PROCESS_INPUT_TIMEOUT_MS, "stdin", http_client=http_client, ) - async def send_signal(pid: int, signal: Literal["terminate", "kill"]) -> None: + async def send_signal(signal: Literal["terminate", "kill"]) -> None: + signal_uuid = _canonical_uuid_key() await self._execute_process_control_rpc( sandbox_id, - build_command_session_send_signal_request(pid, signal), + build_command_session_send_signal_request( + session_uuid=session_uuid, signal=signal, signal_uuid=signal_uuid + ), COMMAND_SESSION_SEND_SIGNAL_RPC_METHOD, _PROCESS_SIGNAL_TIMEOUT_MS, "signal", http_client=http_client, ) - async def reconnect(pid: int | None) -> AsyncIterator[Message]: + async def reconnect(started: bool) -> AsyncIterator[Message]: + # Before a StartEvent it is unknown whether the process was ever + # spawned, so retry Start (create-or-attach); afterwards Connect + # re-attaches, replaying the EndEvent if the process has exited. + # `started` is fixed for this invocation, so pick once. + if started: + method = COMMAND_SESSION_CONNECT_RPC_METHOD + reattach_request = build_command_session_connect_request(session_uuid=session_uuid) + else: + method = COMMAND_SESSION_START_RPC_METHOD + reattach_request = request reauthed = False while True: auth = await self._auth_cache.get_or_refresh(sandbox_id) @@ -2747,19 +2779,9 @@ async def reconnect(pid: int | None) -> AsyncIterator[Message]: http_client=http_client, ) try: - if pid is None: - sessions = await client.execute_unary( - request=build_command_session_list_request(), - method=COMMAND_SESSION_LIST_RPC_METHOD, - headers={"Authorization": f"Bearer {auth['token']}"}, - timeout_ms=_PROCESS_DISCOVERY_TIMEOUT_MS, - ) - pid = find_command_session_pid(sessions, process_tag) - if pid is None: - raise ConnectError(Code.NOT_FOUND, "live process not found") stream = client.execute_server_stream( - request=build_command_session_connect_request(pid), - method=COMMAND_SESSION_CONNECT_RPC_METHOD, + request=reattach_request, + method=method, headers={"Authorization": f"Bearer {auth['token']}"}, timeout_ms=_LIVE_PROCESS_TIMEOUT_MS, ) @@ -2794,8 +2816,19 @@ async def _execute_process_control_rpc( operation: str, http_client: Optional[HTTPClient] = None, ) -> None: - """Run one live-process control RPC with current sandbox auth.""" + """Run one live-process control RPC with current sandbox auth. + + Transient faults are retried with backoff. Retries are safe because the + caller's request carries the operation's idempotency key (input_uuid or + signal_uuid) and is sent byte-identically on every attempt — do not + rebuild it here: sandboxd acknowledges a duplicated apply of the same + key without repeating it. + """ reauthed = False + failures = 0 + # Not a tenacity policy like _gateway_retry: the one-shot 401 reauth + # must retry without consuming a transient attempt, which a decorator's + # single stop counter cannot express. while True: auth = await self._auth_cache.get_or_refresh(sandbox_id) gateway_url = auth["gateway_url"].rstrip("/") @@ -2808,6 +2841,7 @@ async def _execute_process_control_rpc( http_client=http_client, ) try: + # Every attempt resends `request` unchanged; only auth refreshes. await rpc_client.execute_unary( request=request, method=method, @@ -2821,6 +2855,10 @@ async def _execute_process_control_rpc( ): reauthed = True continue + failures += 1 + if failures < _PROCESS_CONTROL_RPC_ATTEMPTS and is_transient_control_fault(error): + await asyncio.sleep(_PROCESS_CONTROL_RETRY_INITIAL_DELAY * 2 ** (failures - 1)) + continue raise APIError( f"process {operation} RPC failed ({error.code.value}): {error.message}" ) from error @@ -2836,7 +2874,9 @@ async def _execute_command_connect_rpc( timeout: Optional[int] = None, ) -> CommandResponse: effective_timeout = timeout if timeout is not None else 300 - request = build_command_session_start_request(command, working_dir, env) + request = build_command_session_start_request( + command=command, working_dir=working_dir, env=env + ) reauthed = False while True: diff --git a/packages/prime-sandboxes/tests/conftest.py b/packages/prime-sandboxes/tests/conftest.py index 6380a75bd..703e576f9 100644 --- a/packages/prime-sandboxes/tests/conftest.py +++ b/packages/prime-sandboxes/tests/conftest.py @@ -1,12 +1,49 @@ -"""Shared pytest configuration and fixtures for sandbox tests""" +"""Shared pytest configuration, fixtures, and gateway RPC test doubles.""" import os import uuid +from datetime import datetime, timedelta, timezone from unittest.mock import patch import pytest from prime_sandboxes import APIClient, SandboxClient +from prime_sandboxes._proto.command_session import command_session_pb2 as pb + +_EV = pb.CommandSessionEvent + + +def _auth_payload(): + """Gateway auth payload returned by the fake auth cache.""" + return { + "gateway_url": "https://gateway.example.com", + "user_ns": "ns", + "job_id": "job", + "token": "tok", + "expires_at": (datetime.now(timezone.utc) + timedelta(minutes=30)).isoformat(), + } + + +class _AsyncFakeCache: + """Always-VM async auth-cache double for gateway RPC tests.""" + + async def get_or_refresh(self, _sandbox_id: str): + return _auth_payload() + + async def is_vm(self, _sandbox_id: str) -> bool: + return True + + +def _start_event(pid, response_type=pb.StartResponse): + return response_type(event=_EV(start=_EV.StartEvent(pid=pid))) + + +def _stdout_event(data, response_type=pb.StartResponse): + return response_type(event=_EV(data=_EV.DataEvent(stdout=data))) + + +def _end_event(code, response_type=pb.StartResponse): + return response_type(event=_EV(end=_EV.EndEvent(exit_code=code))) @pytest.fixture(scope="session") diff --git a/packages/prime-sandboxes/tests/connectrpc_install_smoke.py b/packages/prime-sandboxes/tests/connectrpc_install_smoke.py index dfa20306b..af3eec960 100644 --- a/packages/prime-sandboxes/tests/connectrpc_install_smoke.py +++ b/packages/prime-sandboxes/tests/connectrpc_install_smoke.py @@ -5,7 +5,7 @@ from prime_sandboxes._connectrpc import GOOGLE_PROTOBUF_BINARY_CODEC from prime_sandboxes.rpc_command_session import build_command_session_start_request -request = build_command_session_start_request("echo ready", None, None) +request = build_command_session_start_request(command="echo ready", working_dir=None, env=None) payload = GOOGLE_PROTOBUF_BINARY_CODEC.encode(request) decoded = GOOGLE_PROTOBUF_BINARY_CODEC.decode(payload, type(request)) diff --git a/packages/prime-sandboxes/tests/test_command_transport_selection.py b/packages/prime-sandboxes/tests/test_command_transport_selection.py index 79182de52..50fd1ffe0 100644 --- a/packages/prime-sandboxes/tests/test_command_transport_selection.py +++ b/packages/prime-sandboxes/tests/test_command_transport_selection.py @@ -234,9 +234,12 @@ async def close(self): assert client_init_kwargs["send_compression"] is None assert calls[0][1].command.cwd == "/workspace" assert calls[0][1].command.envs == {"KEY": "value"} - assert calls[1][1].session.pid == 42 + session_uuid = calls[0][1].session_uuid + assert session_uuid # Start carries the client-generated idempotency key + assert calls[1][1].session.session_uuid == session_uuid assert calls[1][1].input.stdin == b"input\n" - assert calls[2][1].session.pid == 42 + assert calls[1][1].input_uuid # each stdin write carries its own idempotency key + assert calls[2][1].session.session_uuid == session_uuid assert calls[2][1].signal == command_session_pb2.SIGNAL_SIGTERM finally: await client.aclose() @@ -254,10 +257,9 @@ async def test_async_open_process_rejects_container_sandbox(): @pytest.mark.asyncio -async def test_process_reconnect_discovers_pid_by_tag_and_refreshes_rejected_auth(monkeypatch): - discovery_tokens = [] - connected_pids = [] - process_tags = [] +async def test_process_recovery_retries_start_and_refreshes_rejected_auth(monkeypatch): + start_requests = [] + retry_tokens = [] class _RejectedTokenCache: def __init__(self): @@ -279,21 +281,24 @@ def __init__(self, _address: str, **_kwargs): pass def execute_server_stream(self, **kwargs): - method = kwargs["method"].name - if method == "Start": - process_tags.append(kwargs["request"].tag) - else: - connected_pids.append(kwargs["request"].session.pid) + assert kwargs["method"].name == "Start" + start_requests.append(kwargs["request"]) + attempt = len(start_requests) + token = kwargs["headers"]["Authorization"] + if attempt > 1: + retry_tokens.append(token) async def events(): - if method == "Start": + if attempt == 1: raise ConnectError(Code.UNAVAILABLE, "stream dropped") - yield command_session_pb2.ConnectResponse( + if token == "Bearer stale": + raise ConnectError(Code.UNAUTHENTICATED, "expired token") + yield command_session_pb2.StartResponse( event=command_session_pb2.CommandSessionEvent( start=command_session_pb2.CommandSessionEvent.StartEvent(pid=42) ) ) - yield command_session_pb2.ConnectResponse( + yield command_session_pb2.StartResponse( event=command_session_pb2.CommandSessionEvent( end=command_session_pb2.CommandSessionEvent.EndEvent(exit_code=0) ) @@ -301,18 +306,6 @@ async def events(): return events() - async def execute_unary(self, **kwargs): - token = kwargs["headers"]["Authorization"] - discovery_tokens.append(token) - if token == "Bearer stale": - raise ConnectError(Code.UNAUTHENTICATED, "expired token") - return command_session_pb2.ListResponse( - sessions=[ - command_session_pb2.CommandSessionInfo(pid=7, tag="other-process"), - command_session_pb2.CommandSessionInfo(pid=42, tag=process_tags[0]), - ] - ) - async def close(self): pass @@ -326,10 +319,12 @@ async def close(self): process = await client.open_process("sbx-vm", "sleep 1") assert await process.wait() == 0 - assert len(process_tags) == 1 - assert process_tags[0].startswith("prime-sdk-") - assert discovery_tokens == ["Bearer stale", "Bearer fresh"] - assert connected_pids == [42] + # The PID was never observed, so recovery re-issues Start; the shared + # session_uuid turns the retries into create-or-attach instead of respawns. + assert len(start_requests) == 3 + assert len({request.session_uuid for request in start_requests}) == 1 + assert start_requests[0].session_uuid + assert retry_tokens == ["Bearer stale", "Bearer fresh"] assert cache.invalidations == 1 await process.aclose() finally: diff --git a/packages/prime-sandboxes/tests/test_connectrpc_compatibility.py b/packages/prime-sandboxes/tests/test_connectrpc_compatibility.py index 1a4cb2e59..c68cf7239 100644 --- a/packages/prime-sandboxes/tests/test_connectrpc_compatibility.py +++ b/packages/prime-sandboxes/tests/test_connectrpc_compatibility.py @@ -10,7 +10,7 @@ def test_google_protobuf_command_request_round_trips() -> None: - request = build_command_session_start_request("echo ready", None, None) + request = build_command_session_start_request(command="echo ready", working_dir=None, env=None) payload = GOOGLE_PROTOBUF_BINARY_CODEC.encode(request) decoded = GOOGLE_PROTOBUF_BINARY_CODEC.decode(payload, type(request)) diff --git a/packages/prime-sandboxes/tests/test_live_process_idempotency.py b/packages/prime-sandboxes/tests/test_live_process_idempotency.py new file mode 100644 index 000000000..560064155 --- /dev/null +++ b/packages/prime-sandboxes/tests/test_live_process_idempotency.py @@ -0,0 +1,160 @@ +"""Live-process RPCs retry safely via sandboxd's idempotency keys. + +Start carries a session_uuid (create-or-attach; see also the Start-retry test in +test_command_transport_selection.py), stdin writes carry an input_uuid +(duplicate applies are acknowledged, not repeated), and signals address the +session by id and carry a signal_uuid (duplicate deliveries are acknowledged, +not repeated), so every transient link fault is retried instead of surfacing. +""" + +import asyncio +from typing import Any, cast + +import pytest +from conftest import _AsyncFakeCache, _end_event, _start_event, _stdout_event +from connectrpc.code import Code +from connectrpc.errors import ConnectError + +from prime_sandboxes._proto.command_session import command_session_pb2 as pb +from prime_sandboxes.core import APIError +from prime_sandboxes.sandbox import AsyncSandboxClient + + +async def _open_process(monkeypatch, fake_client_factory, command="cat"): + monkeypatch.setattr("prime_sandboxes.process._STREAM_RECONNECT_BACKOFF_SECONDS", 0) + monkeypatch.setattr("prime_sandboxes.sandbox._PROCESS_CONTROL_RETRY_INITIAL_DELAY", 0) + monkeypatch.setattr("prime_sandboxes.sandbox.ConnectClient", fake_client_factory) + client = AsyncSandboxClient(api_key="test-key") + cast(Any, client)._auth_cache = _AsyncFakeCache() + return client, await client.open_process("sbx-vm", command) + + +@pytest.mark.asyncio +async def test_reconnect_selects_session_uuid_and_replays_exit(monkeypatch): + """A post-PID drop re-attaches by session selector; a missed exit is replayed.""" + connect_selectors = [] + session_uuids = [] + + class _FakeConnectClient: + def __init__(self, _address: str, **_kwargs): + pass + + def execute_server_stream(self, **kwargs): + method = kwargs["method"].name + + async def events(): + if method == "Start": + session_uuids.append(kwargs["request"].session_uuid) + yield _start_event(42) + yield _stdout_event(b"partial\n") + raise ConnectError(Code.UNAVAILABLE, "stream dropped") + selector = kwargs["request"].session + connect_selectors.append((selector.WhichOneof("selector"), selector.session_uuid)) + # The process exited while detached; Connect replays the retained + # start and end events, mirroring sandboxd since the Connect + # retained-exit fix in platform PR #4735 (same source as Start replay). + yield _start_event(42, pb.ConnectResponse) + yield _end_event(5, pb.ConnectResponse) + + return events() + + async def close(self): + pass + + client, process = await _open_process(monkeypatch, _FakeConnectClient) + try: + assert await process.wait() == 5 + assert connect_selectors == [("session_uuid", session_uuids[0])] + await process.aclose() + finally: + await client.aclose() + + +class _ControlFakeConnectClient: + """Start succeeds; each queued fault is raised by one unary control RPC.""" + + def __init__(self): + self.input_requests = [] + self.signal_requests = [] + self.unary_faults = [] + self.exit_requested = asyncio.Event() + + def execute_server_stream(self, **kwargs): + assert kwargs["method"].name == "Start" + + async def events(): + yield _start_event(42) + await self.exit_requested.wait() + yield _end_event(0) + + return events() + + async def execute_unary(self, **kwargs): + request = kwargs["request"] + if kwargs["method"].name == "SendInput": + self.input_requests.append(request) + if self.unary_faults: + raise self.unary_faults.pop(0) # response lost after the write applied + return pb.SendInputResponse() + assert kwargs["method"].name == "SendSignal" + self.signal_requests.append(request) + if self.unary_faults: + raise self.unary_faults.pop(0) + self.exit_requested.set() + return pb.SendSignalResponse() + + async def close(self): + pass + + +@pytest.mark.asyncio +async def test_stdin_retry_reuses_input_uuid_and_signal_retry_succeeds(monkeypatch): + fake = _ControlFakeConnectClient() + # INTERNAL "Error reading content" exercises the message-marker arm of the + # transient classifier; DEADLINE_EXCEEDED below exercises the code arm. + fake.unary_faults = [ConnectError(Code.INTERNAL, "Error reading content")] + + client, process = await _open_process(monkeypatch, lambda *_args, **_kwargs: fake) + try: + await process.write_stdin(b"hello\n") + + # The write's fault was ambiguous; the retry reused the input_uuid, so the + # server could acknowledge a duplicate apply without writing again. + assert len(fake.input_requests) == 2 + assert fake.input_requests[0].input_uuid + assert fake.input_requests[0].input_uuid == fake.input_requests[1].input_uuid + + await process.write_stdin(b"world\n") + input_uuids = {request.input_uuid for request in fake.input_requests} + assert len(input_uuids) == 2 # a distinct logical write gets a distinct id + + fake.unary_faults = [ConnectError(Code.DEADLINE_EXCEEDED, "deadline")] + await process.terminate() + assert len(fake.signal_requests) == 2 # transient fault, retried + # The retried signal reused the signal_uuid, so the server could + # acknowledge a duplicate delivery without signaling again. + assert fake.signal_requests[0].signal_uuid + assert fake.signal_requests[0].signal_uuid == fake.signal_requests[1].signal_uuid + + assert await process.wait() == 0 + await process.aclose() + finally: + await client.aclose() + + +@pytest.mark.asyncio +async def test_failed_precondition_control_rpc_is_not_retried(monkeypatch): + fake = _ControlFakeConnectClient() + fake.unary_faults = [ConnectError(Code.FAILED_PRECONDITION, "session spec conflict")] + + client, process = await _open_process(monkeypatch, lambda *_args, **_kwargs: fake) + try: + with pytest.raises(APIError, match="failed_precondition"): + await process.write_stdin(b"hello\n") + assert len(fake.input_requests) == 1 # surfaced immediately, no retry + + await process.terminate() + assert await process.wait() == 0 + await process.aclose() + finally: + await client.aclose() diff --git a/packages/prime-sandboxes/tests/test_live_process_idempotency_live.py b/packages/prime-sandboxes/tests/test_live_process_idempotency_live.py new file mode 100644 index 000000000..b73fe710f --- /dev/null +++ b/packages/prime-sandboxes/tests/test_live_process_idempotency_live.py @@ -0,0 +1,233 @@ +"""Live idempotency semantics of VM sandbox live-process RPCs. + +Opt-in E2E tests against a real stack. The hermetic suite proves the client +sends the idempotency keys (session_uuid / input_uuid / signal_uuid); these +tests prove sandboxd actually deduplicates on them: a retried Start attaches +instead of respawning, duplicate SendInput/SendSignal applies land once, and a +Start replay after exit returns the retained EndEvent. + +Enable with PRIME_LIVE_VM_SMOKE=1; PRIME_VM_IMAGE picks the sandbox image. +""" + +import asyncio +import contextlib +import os +import time +import uuid +from unittest import mock + +import pytest +import pytest_asyncio +from connectrpc.client import ConnectClient + +import prime_sandboxes.sandbox as sandbox_module +from prime_sandboxes import AsyncSandboxClient, CreateSandboxRequest +from prime_sandboxes._connectrpc import GOOGLE_PROTOBUF_BINARY_CODEC +from prime_sandboxes.rpc_command_session import ( + COMMAND_SESSION_SEND_INPUT_RPC_METHOD, + COMMAND_SESSION_SEND_SIGNAL_RPC_METHOD, + COMMAND_SESSION_START_RPC_METHOD, + build_command_session_send_input_request, + build_command_session_send_signal_request, + build_command_session_start_request, + parse_command_session_start_event, +) + +pytestmark = [ + pytest.mark.skipif( + os.environ.get("PRIME_LIVE_VM_SMOKE") != "1", + reason="Live VM idempotency smoke is opt-in.", + ), + pytest.mark.asyncio(loop_scope="module"), +] + + +@pytest_asyncio.fixture(scope="module", loop_scope="module") +async def client(): + # No home-dir isolation patch (unlike the platform VM E2E conftest): these + # tests are driven by PRIME_API_KEY / PRIME_API_BASE_URL / PRIME_TEAM_ID, + # which override ~/.prime config; patching Path.home would only hide a + # missing env var by silently falling back to an empty config. + async with AsyncSandboxClient() as client: + yield client + + +@pytest_asyncio.fixture(scope="module", loop_scope="module") +async def vm(client: AsyncSandboxClient): + """One shared RUNNING VM for the whole module, to bound cost.""" + sandbox = await client.create( + CreateSandboxRequest( + name=f"live-idem-{uuid.uuid4().hex[:8]}", + docker_image=os.environ.get("PRIME_VM_IMAGE", "python:3.11-slim"), + vm=True, + cpu_cores=1, + memory_gb=2, + disk_size_gb=10, + timeout_minutes=60, + ) + ) + try: + await client.wait_for_creation(sandbox.id, max_attempts=120) + yield sandbox.id + finally: + await client.delete(sandbox.id) + + +async def _open_process_recording_session_uuid(client, sandbox_id, command): + """Open a live process and capture its session_uuid. + + open_process mints the session_uuid in its body and never exposes it; + recording the mint is the only way a test can rebuild byte-identical + requests that address the live session. + """ + minted: list[str] = [] + real_mint = sandbox_module._canonical_uuid_key + + def recording_mint() -> str: + minted.append(real_mint()) + return minted[-1] + + with mock.patch.object(sandbox_module, "_canonical_uuid_key", recording_mint): + process = await client.open_process(sandbox_id, command) + return process, minted[0] + + +async def _reissue_start(client, sandbox_id, request, *, stop_after_start=False): + """Re-issue a built StartRequest on its own stream, returning parsed events. + + This is the byte-identical retry a pre-StartEvent drop would send; the + stream setup mirrors open_process (auth cache + gateway URL are client + internals with no public single-RPC surface). + """ + auth = await client._auth_cache.get_or_refresh(sandbox_id) + base_url = f"{auth['gateway_url'].rstrip('/')}/{auth['user_ns']}/{auth['job_id']}" + rpc_client = ConnectClient(base_url, codec=GOOGLE_PROTOBUF_BINARY_CODEC, send_compression=None) + stream = rpc_client.execute_server_stream( + request=request, + method=COMMAND_SESSION_START_RPC_METHOD, + headers={"Authorization": f"Bearer {auth['token']}"}, + timeout_ms=60_000, + ) + events = [] + try: + async for response in stream: + event = parse_command_session_start_event(response) + if event is not None: + events.append(event) + if stop_after_start and event is not None and event[0] == "start": + break + finally: + # ConnectClient.close() only flags the client; aclose() is what tears + # down a stream abandoned by stop_after_start deterministically. + with contextlib.suppress(BaseException): + await stream.aclose() + await rpc_client.close() + return events + + +async def _line_count(client, sandbox_id, path) -> int: + result = await client.execute_command(sandbox_id, f"cat {path} 2>/dev/null | wc -l") + return int(result.stdout.strip()) + + +async def _settled_line_count(client, sandbox_id, path) -> int: + """Wait for the first line, then give a duplicate apply time to land.""" + deadline = time.monotonic() + 60 + while time.monotonic() < deadline and await _line_count(client, sandbox_id, path) < 1: + await asyncio.sleep(0.5) + await asyncio.sleep(2) + return await _line_count(client, sandbox_id, path) + + +async def test_start_retry_attaches_not_respawns(client, vm): + """A retried Start against a running session attaches instead of respawning.""" + marker = f"/tmp/live-a-{uuid.uuid4().hex[:8]}" + command = f"echo spawned >> {marker}; exec cat" + process, session_uuid = await _open_process_recording_session_uuid(client, vm, command) + try: + pid = process.pid + request = build_command_session_start_request( + command=command, working_dir=None, env=None, stdin=True, session_uuid=session_uuid + ) + events = await _reissue_start(client, vm, request, stop_after_start=True) + assert events == [("start", pid)], "retried Start spawned a new process" + probe = await client.execute_command( + vm, f"wc -l < {marker}; test -d /proc/{pid} && echo alive" + ) + assert probe.stdout.split() == ["1", "alive"], "retried Start respawned or killed cat" + finally: + await process.aclose() + + +async def test_input_uuid_duplicate_single_write(client, vm): + """A byte-identical SendInput resend (same input_uuid) is applied once.""" + out = f"/tmp/live-b-{uuid.uuid4().hex[:8]}" + process, session_uuid = await _open_process_recording_session_uuid(client, vm, f"cat >> {out}") + try: + request = build_command_session_send_input_request( + session_uuid=session_uuid, + data=b"once\n", + input_uuid=str(uuid.uuid4()), + ) + # _execute_process_control_rpc is the SDK's retry executor; sending one + # built request twice reproduces a retried write byte-for-byte. + for _ in range(2): + await client._execute_process_control_rpc( + vm, + request, + COMMAND_SESSION_SEND_INPUT_RPC_METHOD, + sandbox_module._PROCESS_INPUT_TIMEOUT_MS, + "stdin", + ) + assert await _settled_line_count(client, vm, out) == 1 + finally: + await process.aclose() + + +async def test_signal_uuid_duplicate_single_delivery(client, vm): + """A duplicated SendSignal (same signal_uuid) is delivered once.""" + trap_log = f"/tmp/live-c-{uuid.uuid4().hex[:8]}" + process, session_uuid = await _open_process_recording_session_uuid( + client, vm, f"trap 'echo term >> {trap_log}' TERM; while true; do sleep 0.2; done" + ) + try: + request = build_command_session_send_signal_request( + session_uuid=session_uuid, + signal="terminate", + signal_uuid=str(uuid.uuid4()), + ) + # Same seam as above: the SDK's own retries reuse one signal_uuid, so a + # forced duplicate must resend the identical built request. + for _ in range(2): + await client._execute_process_control_rpc( + vm, + request, + COMMAND_SESSION_SEND_SIGNAL_RPC_METHOD, + sandbox_module._PROCESS_SIGNAL_TIMEOUT_MS, + "signal", + ) + assert await _settled_line_count(client, vm, trap_log) == 1 + finally: + # kill() first: the command traps TERM without exiting, so aclose()'s + # terminate-then-wait escalation would stall on its grace period. + await process.kill() + await process.aclose() + + +async def test_replay_after_exit(client, vm): + """A Start replay after exit returns the retained EndEvent, runs nothing.""" + marker = f"/tmp/live-d-{uuid.uuid4().hex[:8]}" + command = f"echo ran >> {marker}; exit 7" + process, session_uuid = await _open_process_recording_session_uuid(client, vm, command) + try: + assert await asyncio.wait_for(process.wait(), timeout=30) == 7 + finally: + await process.aclose() + + request = build_command_session_start_request( + command=command, working_dir=None, env=None, stdin=True, session_uuid=session_uuid + ) + events = await _reissue_start(client, vm, request) + end_codes = [value for kind, value in events if kind == "end"] + assert end_codes == [7], f"replay did not return the retained exit: {events}" + assert await _line_count(client, vm, marker) == 1, "replayed Start re-ran the command" diff --git a/packages/prime-sandboxes/tests/test_process_stream_reconnect.py b/packages/prime-sandboxes/tests/test_process_stream_reconnect.py index 0d759ca0b..1ba94b2a8 100644 --- a/packages/prime-sandboxes/tests/test_process_stream_reconnect.py +++ b/packages/prime-sandboxes/tests/test_process_stream_reconnect.py @@ -3,38 +3,24 @@ import asyncio import pytest +from conftest import _end_event, _start_event, _stdout_event from connectrpc.code import Code from connectrpc.errors import ConnectError -from prime_sandboxes._proto.command_session import command_session_pb2 as pb from prime_sandboxes.core import APIError from prime_sandboxes.process import AsyncSandboxProcess -_EV = pb.CommandSessionEvent - - -def _start(pid): - return pb.StartResponse(event=_EV(start=_EV.StartEvent(pid=pid))) - - -def _stdout(data): - return pb.StartResponse(event=_EV(data=_EV.DataEvent(stdout=data))) - - -def _end(code): - return pb.StartResponse(event=_EV(end=_EV.EndEvent(exit_code=code))) - class _FakeStreamClient: async def close(self): pass -async def _noop_stdin(pid, data): +async def _noop_stdin(data): pass -async def _noop_signal(pid, sig): +async def _noop_signal(sig): pass @@ -59,19 +45,19 @@ async def test_stream_reconnects_and_resumes_after_drop(fault, monkeypatch): monkeypatch.setattr("prime_sandboxes.process._STREAM_RECONNECT_BACKOFF_SECONDS", 0) async def faulty(): - yield _start(42) - yield _stdout(b"before\n") + yield _start_event(42) + yield _stdout_event(b"before\n") raise fault async def resumed(): - yield _start(42) # Connect re-announces the pid; already known, ignored - yield _stdout(b"after\n") - yield _end(0) + yield _start_event(42) # Connect re-announces the pid; already known, ignored + yield _stdout_event(b"after\n") + yield _end_event(0) reconnect_calls = [] - def reconnect(pid): - reconnect_calls.append(pid) + def reconnect(started): + reconnect_calls.append(started) return resumed() proc = await AsyncSandboxProcess._create( @@ -80,7 +66,7 @@ def reconnect(pid): stdout = await _drain(proc.stdout) rc = await proc.wait() - assert reconnect_calls == [42] + assert reconnect_calls == [True] assert rc == 0 # exit observed on the resumed stream assert stdout == b"before\nafter\n" # output from both segments await proc.aclose() @@ -91,18 +77,18 @@ async def test_stream_reconnects_after_clean_eof(monkeypatch): monkeypatch.setattr("prime_sandboxes.process._STREAM_RECONNECT_BACKOFF_SECONDS", 0) async def ended_without_exit(): - yield _start(42) - yield _stdout(b"before\n") + yield _start_event(42) + yield _stdout_event(b"before\n") async def resumed(): - yield _start(42) - yield _stdout(b"after\n") - yield _end(0) + yield _start_event(42) + yield _stdout_event(b"after\n") + yield _end_event(0) reconnect_calls = [] - def reconnect(pid): - reconnect_calls.append(pid) + def reconnect(started): + reconnect_calls.append(started) return resumed() proc = await AsyncSandboxProcess._create( @@ -115,28 +101,27 @@ def reconnect(pid): assert await _drain(proc.stdout) == b"before\nafter\n" assert await proc.wait() == 0 - assert reconnect_calls == [42] + assert reconnect_calls == [True] await proc.aclose() @pytest.mark.asyncio -async def test_stream_reconnects_before_pid_is_received(monkeypatch): +async def test_stream_recovers_before_pid_is_received(monkeypatch): monkeypatch.setattr("prime_sandboxes.process._STREAM_RECONNECT_BACKOFF_SECONDS", 0) async def dropped_before_start(): raise ConnectError(Code.UNAVAILABLE, "stream dropped") - yield _start(42) + yield _start_event(42) reconnect_calls = [] - def reconnect(pid): - reconnect_calls.append(pid) + def reconnect(started): + # started=False tells the caller to retry Start (create-or-attach). + reconnect_calls.append(started) async def stream(): - if len(reconnect_calls) == 1: - raise ConnectError(Code.NOT_FOUND, "process not registered yet") - yield _start(42) - yield _end(0) + yield _start_event(42) + yield _end_event(0) return stream() @@ -150,14 +135,41 @@ async def stream(): assert proc.pid == 42 assert await proc.wait() == 0 - assert reconnect_calls == [None, None] + assert reconnect_calls == [False] + await proc.aclose() + + +@pytest.mark.asyncio +async def test_reconnect_after_exit_replays_end_event(monkeypatch): + monkeypatch.setattr("prime_sandboxes.process._STREAM_RECONNECT_BACKOFF_SECONDS", 0) + + async def dropped_mid_stream(): + yield _start_event(42) + yield _stdout_event(b"before\n") + raise ConnectError(Code.UNAVAILABLE, "stream dropped") + + async def replayed_after_exit(): + # sandboxd retains exited sessions briefly; Connect replays the exit. + yield _start_event(42) + yield _end_event(3) + + proc = await AsyncSandboxProcess._create( + _FakeStreamClient(), + dropped_mid_stream(), + _noop_stdin, + _noop_signal, + reconnect=lambda started: replayed_after_exit(), + ) + + assert await _drain(proc.stdout) == b"before\n" + assert await proc.wait() == 3 await proc.aclose() @pytest.mark.asyncio async def test_end_before_pid_fails_instead_of_hanging(): async def ended_before_start(): - yield _end(0) + yield _end_event(0) with pytest.raises(APIError, match="ended before reporting its PID"): await asyncio.wait_for( @@ -175,7 +187,7 @@ async def ended_before_start(): async def test_stream_without_reconnect_still_fails(): # No reconnect callable preserves the previous fatal behavior. async def faulty(): - yield _start(7) + yield _start_event(7) raise ConnectError(Code.UNAVAILABLE, "error reading a body from connection: timed out") proc = await AsyncSandboxProcess._create( @@ -187,15 +199,23 @@ async def faulty(): @pytest.mark.asyncio -async def test_permanent_fault_is_not_reconnected(): +@pytest.mark.parametrize( + "fault", + [ + ConnectError(Code.NOT_FOUND, "session gone"), + ConnectError(Code.FAILED_PRECONDITION, "session spec conflict"), + ], + ids=["not_found", "failed_precondition"], +) +async def test_permanent_fault_is_not_reconnected(fault): async def faulty(): - yield _start(9) - raise ConnectError(Code.NOT_FOUND, "session gone") + yield _start_event(9) + raise fault calls = [] - def reconnect(pid): - calls.append(pid) + def reconnect(started): + calls.append(started) raise AssertionError("should not reconnect on a permanent fault") proc = await AsyncSandboxProcess._create(