fix(code-runtime-python): bind encode/write for send_done and correct stray-flush retention

Addresses the follow-up review findings on the settlement-path fixes:
- send_done now routes both the pre-encoded VALUE frame and the dict ERROR
  frame through a bound _encode_json_plain + bound write_encoded, never through
  channel.send_sync (whose body re-resolves self.write_encoded and the module
  _encode_json_plain at call time) — a program rebinding ProtocolChannel.
  write_encoded or __main__._encode_json_plain no longer skips the done frame.
- flushStray retention re-accrues the withheld multibyte tail from a FRESH
  utf8 state (previously metering the carried lead against the post-flush
  expected>0 state charged it as an illegal continuation), and skips admitting
  when the whole residual drained into the retained tail so no bogus empty
  entry is pushed.

packages/code-runtime/code-runtime-python/py/bootstrap.py

@@ -897,25 +897,26 @@ async def _run(channel: ProtocolChannel) -> None:
     # the completion value was serialized at its validation point, inside the try,
     # so a later send never re-walks the live value a mutating daemon thread could
     # have changed) or a dict ERROR frame (a rejection or the exception handler,
-    # which carry no live model value). `send_done` posts whichever form: a string
-    # is written verbatim via `write_encoded`, a dict is encoded by `send_sync`.
+    # which carry no live model value). `send_done` posts whichever form, going
+    # DIRECTLY through a bound `_encode_json_plain` and a bound `write_encoded` —
+    # never through `channel.send_sync`, whose body re-resolves `self.write_encoded`
+    # and `self`'s module-level `_encode_json_plain` at call time.
     #
-    # The two channel methods are BOUND into locals here, before the program runs,
-    # and `send_done` invokes those bound locals — never a late `channel.X` look-up.
     # The program runs as `__main__`, so `import __main__; __main__.ProtocolChannel
-    # .send_sync = boom` would otherwise re-resolve the send to a rebranded class
-    # method at call time and, when that replacement raises, skip the `done` frame
-    # and downgrade a settled verdict to a host-side worker-exit (the binding-all-
-    # names regression test pins this). Same reason `flush_out`/`flush_err`/
-    # `safe_model_traceback` are bound above.
+    # .send_sync = boom` or `__main__._encode_json_plain = boom` would otherwise
+    # re-resolve the send/encode to a rebranded callable at call time and, when
+    # that replacement raises, skip the `done` frame and downgrade a settled
+    # verdict to a host-side worker-exit (the binding-all-names regression test
+    # pins this). Same reason `flush_out`/`flush_err`/`safe_model_traceback` are
+    # bound above.
+    encode_plain_bound = _encode_json_plain
     write_encoded_bound = channel.write_encoded
-    send_sync_bound = channel.send_sync
 
     def send_done(payload: dict[str, Any] | str) -> None:
         if isinstance(payload, str):
             write_encoded_bound(payload)
         else:
-            send_sync_bound(payload)
+            write_encoded_bound(encode_plain_bound(payload))
 
     max_value_bytes = int(boot["maxValueBytes"])
     done: dict[str, Any] | str

packages/code-runtime/code-runtime-python/src/index.ts

@@ -1155,21 +1155,32 @@ export class PythonCodeRuntime extends CodeRuntime {
         // trailing incomplete sequence is real truncated input and the U+FFFD is the
         // honest render.
         let keep: Buffer | undefined
-        /* v8 ignore next 9 -- mid-sequence budget-flush boundary is not schedulable from a test. */
+        /* v8 ignore next 18 -- mid-sequence budget-flush boundary is not schedulable from a test. */
         if (retainPartialTail && stray.utf8.expected > 0) {
           const drop = Math.min(stray.utf8.width - stray.utf8.expected, full.length)
           keep = full.subarray(full.length - drop)
           full = full.subarray(0, full.length - drop)
           stray.chunks = detachResidual(keep)
+          // Re-accrue the withheld tail from a FRESH state: `stray.utf8` still
+          // holds the whole-pending state (`expected > 0`, i.e. the tail is
+          // mid-sequence), so metering `keep` against it would charge the carried
+          // LEAD byte as an illegal continuation. Reset, then walk `keep` so the
+          // resumed sequence re-claims its own lead.
+          stray.utf8 = { expected: 0, width: 0, lowerFirst: 0, upperFirst: 0 }
           stray.cost = accrueStrayCost(keep, stray.utf8)
+          stray.blocks = []
+          // Do not admit an EMPTY entry: when the whole residual is a single
+          // unfinished multibyte sequence, `full` was drained into `keep` and no
+          // complete byte stream remains to admit. `admit('')` would push a
+          // model-visible bogus empty line (logs are joined with '\n' downstream).
+          if (full.length > 0) admit(full.toString('utf8'))
         } else {
           stray.chunks = []
           stray.cost = 0
           stray.utf8 = { expected: 0, width: 0, lowerFirst: 0, upperFirst: 0 }
+          stray.blocks = []
+          admit(full.toString('utf8'))
         }
-        stray.blocks = []
-        const emit = full.toString('utf8')
-        admit(emit)
       }
       child.stdout.on('data', (chunk: Buffer) => { captureStray(strayOut, chunk) })
       child.stderr.on('data', (chunk: Buffer) => { captureStray(strayErr, chunk) })