test(subagent-sdk): make the partial-output test deterministic

The aborted-mid-stream case raced the abort against the child's chunk
delivery across two pipes and lost under full-suite load. Replace it with a
same-pipe ordering probe: the fake streams one text-delta chunk and then
answers the prompt with a malformed (non-accepted) result, so frame order
guarantees the chunk precedes the failure — the accumulated partial text
must survive into the error result. Same code path (collectOutput without a
complete assistant/message), no timing window.

packages/sdk/sdk-client/tests/fake-runtime.ts

@@ -21,9 +21,9 @@
  *   arrives, then poll for the GO file before answering (deterministic
  *   cancel-during-handshake window).
  * - `FAKE_HANG_PROMPT`: never answer `session/prompt` (for timeout/dispose tests).
- * - `FAKE_STREAM_THEN_HANG`: stream a text chunk for the prompt, then never
- *   finish the turn or answer (partial-output cancel probe). Touches
- *   `FAKE_STREAM_READY` after the chunk when set.
+ * - `FAKE_STREAM_THEN_MALFORMED`: stream a text chunk for the prompt, then
+ *   answer `{}` (no accepted) — same-pipe ordering makes the chunk arrive
+ *   before the protocol failure (partial-output retention probe).
  * - `FAKE_IGNORE_EOF` + `FAKE_SIGTERM_FILE`: keep running after stdin EOF; touch the file on SIGTERM (ladder probe).
  * - `FAKE_TRAP_SIGTERM`: with `FAKE_IGNORE_EOF`, survive SIGTERM too (SIGKILL-rung probe).
  * - `FAKE_EXIT_BEFORE_INIT`: exit 3 immediately (spawn-then-die probe).
@@ -151,10 +151,10 @@ reader.on('line', (line) => {
       respond({ serverInfo: { name: 'deepseek-harness-sdk-runtime', version: '0.0.1' } })
       return
     case 'session/prompt': {
-      if (env.FAKE_STREAM_THEN_HANG !== undefined) {
+      if (env.FAKE_STREAM_THEN_MALFORMED !== undefined) {
         const sessionId = sessionIdOf(frame.params)
-        event(sessionId, 'assistant/chunk', { turn: 0, step: 0, chunk: { type: 'text-delta', index: 0, text: 'streamed then hung' } })
-        if (env.FAKE_STREAM_READY !== undefined) writeFileSync(env.FAKE_STREAM_READY, 'streamed\n')
+        event(sessionId, 'assistant/chunk', { turn: 0, step: 0, chunk: { type: 'text-delta', index: 0, text: 'streamed then cut short' } })
+        respond({})
         return
       }
       if (env.FAKE_HANG_PROMPT !== undefined) return

packages/subagent/subagent-sdk/tests/subagent-sdk.spec.ts

@@ -209,27 +209,18 @@ describe('dsh-subagent-sdk provider', () => {
     }
   })
 
-  it('keeps partial streamed text when aborted mid-turn', async () => {
-    const tmp = mkdtempSync(join(tmpdir(), 'subagent-sdk-partial-'))
-    const streamed = join(tmp, 'streamed')
-    try {
-      const ctx = await setup(
-        { FAKE_STREAM_THEN_HANG: '1', FAKE_STREAM_READY: streamed },
-        { disposeEofGraceMs: 200, disposeGraceMs: 200, shutdownTimeoutMs: 100 },
-      )
-      const controller = new AbortController()
-      const run = await ctx.subagents.start('sdk', request('p', controller.signal))
-      // Cancel only after the chunk has demonstrably streamed (condition, not a sleep).
-      await waitForFile(streamed)
-      controller.abort('test')
-      const result = await run.result
-      expect(result.stopReason).toBe('aborted')
-      expect(text(result.output)).toBe('streamed then hung')
-      await run.dispose()
-      await ctx.fiber.dispose()
-    } finally {
-      rmSync(tmp, { recursive: true, force: true })
-    }
+  it('keeps accumulated streamed text when the turn is cut short before a full message', async () => {
+    // The fake streams one text-delta chunk and then violates the protocol on
+    // the same pipe; frame order guarantees the chunk was dispatched before
+    // the failure settles, so the accumulated partial text (no complete
+    // assistant/message ever arrived) must survive into the error result.
+    const ctx = await setup({ FAKE_STREAM_THEN_MALFORMED: '1' }, { shutdownTimeoutMs: 100, disposeEofGraceMs: 200, disposeGraceMs: 200 })
+    const run = await ctx.subagents.start('sdk', request())
+    const result = await run.result
+    expect(result.stopReason).toBe('error')
+    expect(text(result.output)).toBe('streamed then cut short')
+    await run.dispose()
+    await ctx.fiber.dispose()
   })
 
   it('dispose cancels a hung child locally and reaps it', async () => {