more robust message header: prefer explicit props_length/props_chunks over odd YXML.embed_controls;

--- a/src/Pure/General/properties.scala	Fri Jul 05 00:12:32 2024 +0200
+++ b/src/Pure/General/properties.scala	Fri Jul 05 00:18:51 2024 +0200
@@ -21,6 +21,9 @@
       val i = str.indexOf('=')
       if (i <= 0) None else Some((str.substring(0, i), str.substring(i + 1)))
     }
+
+    def parse(s: java.lang.String): Entry =
+      unapply(s) getOrElse error("Bad property entry: " + quote(s))
   }
 
   def defined(props: T, name: java.lang.String): java.lang.Boolean =

--- a/src/Pure/General/value.scala	Fri Jul 05 00:12:32 2024 +0200
+++ b/src/Pure/General/value.scala	Fri Jul 05 00:18:51 2024 +0200
@@ -21,6 +21,9 @@
   }
 
   object Nat {
+    def unapply(bs: Bytes): Option[scala.Int] =
+      if (bs.byte_iterator.forall(b => '0' <= b && b <= '9')) unapply(bs.text)
+      else None
     def unapply(s: java.lang.String): Option[scala.Int] =
       s match {
         case Int(n) if n >= 0 => Some(n)

--- a/src/Pure/PIDE/prover.scala	Fri Jul 05 00:12:32 2024 +0200
+++ b/src/Pure/PIDE/prover.scala	Fri Jul 05 00:18:51 2024 +0200
@@ -246,8 +246,13 @@
   /* message output */
 
   private def message_output(stream: InputStream): Thread = {
-    def decode_chunk(chunk: Bytes): XML.Body =
-      Symbol.decode_yxml_failsafe(chunk.text, cache = cache)
+    def decode_prop(bytes: Bytes): Properties.Entry = {
+      val (a, b) = Properties.Eq.parse(bytes.text)
+      (Symbol.decode(a), Symbol.decode(b))
+    }
+
+    def decode_xml(bytes: Bytes): XML.Body =
+      Symbol.decode_yxml_failsafe(bytes.text, cache = cache)
 
     val thread_name = "message_output"
     Isabelle_Thread.fork(name = thread_name) {
@@ -256,13 +261,12 @@
         while (!finished) {
           Byte_Message.read_message(stream) match {
             case None => finished = true
-            case Some(header :: chunks) =>
-              decode_chunk(header) match {
-                case List(XML.Elem(Markup(kind, props), Nil)) =>
-                  if (kind == Markup.PROTOCOL) protocol_output(props, chunks)
-                  else output(kind, props, decode_chunk(Prover.the_chunk(chunks, kind)))
-                case _ => Prover.bad_header(header.toString)
-              }
+            case Some(k :: Value.Nat(props_length) :: rest) =>
+              val kind = k.text
+              val props = rest.take(props_length).map(decode_prop)
+              val chunks = rest.drop(props_length)
+              if (kind == Markup.PROTOCOL) protocol_output(props, chunks)
+              else output(kind, props, chunks.flatMap(decode_xml))
             case Some(_) => Prover.bad_chunks()
           }
         }

--- a/src/Pure/System/message_channel.ML	Fri Jul 05 00:12:32 2024 +0200
+++ b/src/Pure/System/message_channel.ML	Fri Jul 05 00:18:51 2024 +0200
@@ -24,11 +24,12 @@
   Mailbox.await_empty mbox;
   Isabelle_Thread.join thread);
 
-fun message (Message_Channel {mbox, ...}) name raw_props chunks =
+fun message (Message_Channel {mbox, ...}) name props chunks =
   let
-    val robust_props = map (apply2 YXML.embed_controls) raw_props;
-    val header = [XML.Elem ((name, robust_props), [])];
-  in Mailbox.send mbox (Message (header :: chunks)) end;
+    val kind = XML.Encode.string name;
+    val props_length = XML.Encode.int (length props);
+    val props_chunks = map (XML.Encode.string o Properties.print_eq) props;
+  in Mailbox.send mbox (Message (kind :: props_length :: props_chunks @ chunks)) end;
 
 fun make stream =
   let