fix(code-runtime-python): merge a flushed unterminated line into the next log entry

The review's remaining warning: an explicit flush of an unterminated line
(print(..., end='', flush=True)) pushed a full log frame, so the following
print() landed in a second entry and logs.join('\n') rendered 'a\nb' for what
the program printed as one line — a model-visible output defect. The flush
frame now carries an  flag (LogMessage gains the optional field on both
sides and in the mirror test), the host holds it and appends the next log frame
to the same entry, and finish() admits the residual if the run ends with it
still open. The settlement note registers the decimal-context fix from the
previous commit.

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

@@ -149,11 +149,11 @@ class LogBuffer:
 
         return 0 if self._truncated else self._remaining
 
-    def push(self, text: str) -> None:
+    def push(self, text: str, open: bool = False) -> None:
         with self._lock:
-            self._push_locked(text)
+            self._push_locked(text, open)
 
-    def _push_locked(self, text: str) -> None:
+    def _push_locked(self, text: str, open: bool = False) -> None:
         if self._truncated:
             return
         # Cheap lower bound FIRST: one char is at least one UTF-8 byte and the
@@ -193,7 +193,7 @@ class LogBuffer:
             self._sink(log_truncation_marker(self._max_bytes), truncated=True)
             return
         self._remaining -= cost
-        self._sink(text)
+        self._sink(text, open=open)
 
 
 class _LogStream(io.TextIOBase):
@@ -445,7 +445,10 @@ class _LogStream(io.TextIOBase):
                 self._pending = []
                 self._pending_blocks = []
                 self._pending_chars = 0
-                self._logs.push(line)
+                # The line has NO trailing newline: mark the frame `open` so the
+                # host appends the next log frame to the same entry instead of
+                # inserting a fake newline between two entries.
+                self._logs.push(line, open=True)
 
 
 # ---------------------------------------------------------------------------
@@ -1044,9 +1047,14 @@ async def _run(channel: ProtocolChannel) -> None:
         # The sink writes through the def-time bound encode+write primitives
         # (not _send_sync_cls, whose body still resolves _encode_json_plain and
         # self.write_encoded at call time) so a rebind cannot break a log frame.
-        sink=lambda text, truncated=False: _write_encoded_cls(
+        sink=lambda text, truncated=False, open=False: _write_encoded_cls(
             _encode_plain_cls(
-                {"type": "log", "text": text, **({"truncated": True} if truncated else {})}
+                {
+                    "type": "log",
+                    "text": text,
+                    **({"truncated": True} if truncated else {}),
+                    **({"open": True} if open else {}),
+                }
             )
         ),
     )

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

@@ -81,10 +81,11 @@ class LogMessage(_LogMessageRequired, total=False):
 
     ``truncated`` is set only on the frame that IS the child ledger's truncation
     marker (not program output), so the host stops capturing at the same point
-    the child did — mirrors the TS `truncated?`.
+    the child did — mirrors the TS `truncated?`. ``open`` is set on a flushed unterminated line the host appends the next frame to (mirrors `open?`).
     """
 
     truncated: bool
+    open: bool
 
 
 class DoneErrorField(TypedDict):

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

@@ -1025,6 +1025,10 @@ export class PythonCodeRuntime extends CodeRuntime {
     return new Promise<CodeRunResult>((resolve) => {
       let settled = false
       const logs: string[] = []
+      // An unterminated line flushed with the `open` flag: the next log frame
+      // appends to it (no fake newline between entries), and finish() admits
+      // the residual if the run ends with it still open.
+      let openLog: string | undefined
 
       // One host-side ledger covers normal frames, forged frames, and stray stdout bytes.
       // The ledger starts one byte below maxLogBytes: each entry is charged its
@@ -1496,7 +1500,17 @@ export class PythonCodeRuntime extends CodeRuntime {
               }
               return
             }
-            admit(message.text)
+            if (message.open === true) {
+              // An explicit flush of an unterminated line: hold it so the next
+              // frame appends to the SAME entry (print('a', end='', flush=True)
+              // followed by print('b') reads back as one 'ab' entry, not a fake
+              // newline). The residual is admitted by finish() if the run ends
+              // with it still open.
+              openLog = (openLog ?? '') + message.text
+              return
+            }
+            admit((openLog ?? '') + message.text)
+            openLog = undefined
             return
           case 'done': {
             if (message.error) {
@@ -1876,6 +1890,12 @@ export class PythonCodeRuntime extends CodeRuntime {
         // A spawn failure (ENOENT, EACCES) never produced a pid, so there is no
         // process to kill: settle now. Its `close` still fires later and reaches
         // the idempotent settle() again as a no-op.
+        // An unterminated flushed line never got a closing frame; admit it so
+        // the committed flush is not lost from logs.
+        if (openLog !== undefined) {
+          admit(openLog)
+          openLog = undefined
+        }
         if (child.pid === undefined) {
           settle(result)
           return

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

@@ -102,6 +102,13 @@ interface LogMessage {
    * and keeps exactly one marker in `logs`.
    */
   truncated?: boolean
+  /**
+   * Set on the frame an explicit `flush()` (or the settlement flush) pushes for
+   * an UNTERMINATED line: the host holds it and appends the next log frame to
+   * the same entry, so `print('a', end='', flush=True); print('b')` reads back
+   * as one `'ab'` entry rather than a fake newline between two entries.
+   */
+  open?: boolean
 }
 
 /** The failure carried on a {@link DoneMessage}: one of three kinds plus text. */
@@ -227,7 +234,7 @@ const WIRE_FRAME_FIELD_ROLES = {
   RunMessage: { type: 'required', program: 'required' },
   BootAckMessage: { type: 'required' },
   CallMessage: { type: 'required', id: 'required', global: 'required', name: 'required', args: 'required' },
-  LogMessage: { type: 'required', text: 'required', truncated: 'optional' },
+  LogMessage: { type: 'required', text: 'required', truncated: 'optional', open: 'optional' },
   DoneErrorField: { kind: 'required', message: 'required' },
   DoneMessage: { type: 'required', value: 'optional', error: 'optional' },
   ErrorClass: { name: 'required', memberNameProperty: 'required' },
@@ -611,8 +618,13 @@ export function validateChildFrame(raw: unknown): ChildToHost | undefined {
       if (typeof m.text !== 'string') return undefined
       // Rebuilt, not passed through: a forged `truncated` of any other type
       // would reach the host as a truthy value and silence capture for the
-      // rest of the run. Only the literal `true` counts.
-      return { type: 'log', text: m.text, ...m.truncated === true ? { truncated: true } : {} }
+      // rest of the run. Only the literal `true` counts; `open` likewise.
+      return {
+        type: 'log',
+        text: m.text,
+        ...m.truncated === true ? { truncated: true } : {},
+        ...m.open === true ? { open: true } : {},
+      }
     case 'call': {
       // The id must be a finite number: it is echoed verbatim into the reply
       // frame, and a forged `1e400` id (Infinity after JSON.parse) would make

packages/code-runtime/code-runtime-python/tests/runtime.spec.ts

@@ -1843,6 +1843,41 @@ describe('PythonCodeRuntime — programs and bindings', () => {
     expect(result.error?.kind).not.toBe('worker-exit')
   }, 90_000)
 
+  it('appends a flushed unterminated line to the next entry without a fake newline', async () => {
+    // An explicit flush of an unterminated line (print(..., end='', flush=True))
+    // used to push a full log frame, so the following print() landed in a
+    // SECOND entry and logs.join('\n') rendered 'a\nb' for what the program
+    // printed as one line. The flush frame now carries `open: true` and the
+    // host appends the next frame to the same entry.
+    const { runtime } = await setup()
+    const result = await runtime.run({
+      program: [
+        "print('a', end='', flush=True)",
+        "print('b')",
+        'return "done"',
+      ].join('\n'),
+      bindings: [],
+    })
+    expect(result.error).toBeUndefined()
+    expect(result.logs).toEqual(['ab'])
+  }, 15_000)
+
+  it('keeps a flushed unterminated line when the run ends with it still open', async () => {
+    // The settlement flush pushes the residual with `open: true`; finish()
+    // admits it so a program that commits a partial line and returns does not
+    // lose it from logs.
+    const { runtime } = await setup()
+    const result = await runtime.run({
+      program: [
+        "print('committed', end='', flush=True)",
+        'return "done"',
+      ].join('\n'),
+      bindings: [],
+    })
+    expect(result.error).toBeUndefined()
+    expect(result.logs).toEqual(['committed'])
+  }, 15_000)
+
   it('keeps a float completion exact when the program mutates the decimal context', async () => {
     // The float encoder's Decimal(repr(value)).normalize() used the process
     // GLOBAL decimal context: a legitimate program setting