fix(opencode): wait for stdout writes before exit so piped JSON is not truncated

packages/opencode/src/cli/cmd/db.ts

@@ -4,6 +4,7 @@ import { Database } from "@opencode-ai/core/database/database"
 import { Effect } from "effect"
 import { sql } from "drizzle-orm"
 import { effectCmd } from "../effect-cmd"
+import { writeStdout } from "../stdout"
 
 const QueryCommand = effectCmd({
   command: "$0 [query]",
@@ -27,7 +28,7 @@ const QueryCommand = effectCmd({
     if (query) {
       const { db } = yield* Database.Service
       const result = yield* db.all<Record<string, unknown>>(sql.raw(query)).pipe(Effect.orDie)
-      if (args.format === "json") console.log(JSON.stringify(result, null, 2))
+      if (args.format === "json") yield* writeStdout(JSON.stringify(result, null, 2) + "\n")
       else if (result.length > 0) {
         const keys = Object.keys(result[0])
         console.log(keys.join("\t"))

packages/opencode/src/cli/cmd/export.ts

@@ -3,6 +3,7 @@ import { SessionV1 } from "@opencode-ai/core/v1/session"
 import { MessageV2 } from "../../session/message-v2"
 import { SessionID } from "../../session/schema"
 import { effectCmd, fail } from "../effect-cmd"
+import { writeStdout } from "../stdout"
 import { UI } from "../ui"
 import * as prompts from "@clack/prompts"
 import { EOL } from "os"
@@ -280,13 +281,12 @@ const run = Effect.fn("Cli.export.body")(function* (args: { sessionID?: string;
 
   // Match legacy try/catch — catches both typed failures and defects
   // (Session.Service.get throws NotFoundError as a defect, not a typed E).
-  return yield* Effect.gen(function* () {
+  const exportData = yield* Effect.gen(function* () {
     const sessionInfo = yield* svc.get(sessionID!)
     const messages = yield* svc.messages({ sessionID: sessionInfo.id })
 
-    const exportData = { info: sessionInfo, messages }
-
-    process.stdout.write(JSON.stringify(args.sanitize ? sanitize(exportData) : exportData, null, 2))
-    process.stdout.write(EOL)
+    return { info: sessionInfo, messages }
   }).pipe(Effect.catchCause(() => fail(`Session not found: ${sessionID!}`)))
+
+  yield* writeStdout(JSON.stringify(args.sanitize ? sanitize(exportData) : exportData, null, 2) + EOL)
 })

packages/opencode/src/cli/cmd/session.ts

@@ -2,6 +2,7 @@ import type { Argv } from "yargs"
 import { Effect } from "effect"
 import { cmd } from "./cmd"
 import { effectCmd, fail } from "../effect-cmd"
+import { writeStdout } from "../stdout"
 import { Session } from "@/session/session"
 import { SessionID } from "../../session/schema"
 import { UI } from "../ui"
@@ -110,7 +111,7 @@ export const SessionListCommand = effectCmd({
         await proc.exited
       })
     } else {
-      console.log(output)
+      yield* writeStdout(output + "\n")
     }
   }),
 })

packages/opencode/src/cli/stdout.ts

@@ -0,0 +1,21 @@
+import { Effect } from "effect"
+import { CliError } from "./effect-cmd"
+
+// stdout is asynchronous when it is a pipe, and src/index.ts terminates
+// as soon as a command returns, so anything still queued on the stream is lost
+// (#29330). Wait for the write callback like generate.ts does. EPIPE means the
+// reader went away (`| head`) and counts as done; the 'error' event that follows a
+// failed write keeps its listener so it cannot crash the process.
+export const writeStdout = (text: string) =>
+  Effect.tryPromise({
+    try: () =>
+      new Promise<void>((resolve, reject) => {
+        const settle = (err?: Error | null) => {
+          const code = (err as NodeJS.ErrnoException | null | undefined)?.code
+          return !err || code === "EPIPE" ? resolve() : reject(err)
+        }
+        process.stdout.once("error", settle)
+        process.stdout.write(text, settle)
+      }),
+    catch: (cause) => new CliError({ message: `failed to write to stdout: ${String(cause)}` }),
+  })

packages/opencode/test/cli/stdout.test.ts

@@ -0,0 +1,72 @@
+import { afterEach, beforeEach, expect, test } from "bun:test"
+import { mkdtemp, rm } from "node:fs/promises"
+import { tmpdir } from "node:os"
+import path from "node:path"
+
+const QUERY =
+  "with recursive t(n) as (select 1 union all select n+1 from t where n < 15000) select n as n, '0123456789012345678901234567890123456789' as pad from t"
+const ROOT = path.resolve(import.meta.dir, "../..")
+
+let tmp: string
+
+beforeEach(async () => {
+  tmp = await mkdtemp(path.join(tmpdir(), "opencode-stdout-"))
+})
+
+afterEach(async () => {
+  await rm(tmp, { recursive: true, force: true })
+})
+
+function spawn(stdout: "pipe"): Bun.Subprocess<"ignore", "pipe", "pipe">
+function spawn(stdout: ReturnType<typeof Bun.file>): Bun.Subprocess<"ignore", ReturnType<typeof Bun.file>, "pipe">
+function spawn(stdout: "pipe" | ReturnType<typeof Bun.file>) {
+  const isolated = {
+    OPENCODE_TEST_HOME: tmp,
+    HOME: tmp,
+    XDG_CONFIG_HOME: path.join(tmp, ".config"),
+    XDG_DATA_HOME: path.join(tmp, ".local/share"),
+    XDG_STATE_HOME: path.join(tmp, ".local/state"),
+    XDG_CACHE_HOME: path.join(tmp, ".cache"),
+    OPENCODE_CONFIG_CONTENT: "{}",
+    OPENCODE_DISABLE_PROJECT_CONFIG: "1",
+    OPENCODE_PURE: "1",
+    OPENCODE_DISABLE_AUTOUPDATE: "1",
+    OPENCODE_DISABLE_AUTOCOMPACT: "1",
+    OPENCODE_DISABLE_MODELS_FETCH: "1",
+    OPENCODE_AUTH_CONTENT: "{}",
+  }
+  return Bun.spawn([process.execPath, "src/index.ts", "db", "--format", "json", QUERY], {
+    cwd: ROOT,
+    stdout,
+    stderr: "pipe",
+    env: { ...process.env, ...isolated },
+  })
+}
+
+// Delay the first read until the pipe fills so returning before stdout drains
+// deterministically exposes truncated output.
+test("large JSON output is complete when the reader is slow", async () => {
+  const proc = spawn("pipe")
+  await Bun.sleep(2000)
+  const text = await new Response(proc.stdout).text()
+  await proc.exited
+  expect(proc.exitCode).toBe(0)
+  expect(text.length).toBeGreaterThan(65536)
+  expect(JSON.parse(text)).toHaveLength(15000)
+
+  const file = path.join(tmp, "full.json")
+  const ctl = spawn(Bun.file(file))
+  await ctl.exited
+  expect(text).toBe(await Bun.file(file).text())
+}, 60_000)
+
+test("exits cleanly when the reader goes away", async () => {
+  const proc = spawn("pipe")
+  const reader = proc.stdout.getReader()
+  await reader.read()
+  await reader.cancel()
+  const code = await Promise.race([proc.exited, Bun.sleep(5000).then(() => "timeout")])
+  expect(code).toBe(0)
+  const err = await new Response(proc.stderr).text()
+  expect(err).not.toMatch(/EPIPE|Unhandled/)
+}, 60_000)