merge(dev): resolve instance event bridge conflict

Merge upstream dev at 1d4f778, retaining the shared LOOPBACK_HOST and instance client factory introduced by #630 while preserving per-runtime stream IDs, restart fencing, bounded SSE parsing, and reader cleanup from #626.

Validated with the server typecheck and focused instance-client, background-process, and instance-event tests (13 passing).

Author: pascalandr

packages/server/src/background-processes/manager.test.ts

@@ -0,0 +1,153 @@
+import assert from "node:assert/strict"
+import { describe, it } from "node:test"
+import { promises as fs } from "node:fs"
+import path from "node:path"
+import os from "node:os"
+
+import { BackgroundProcessManager } from "./manager"
+import type { WorkspaceManager } from "../workspaces/manager"
+import type { EventBus } from "../events/bus"
+import type { Logger } from "../logger"
+
+const WORKSPACE_ID = "ws-test"
+const SESSION_ID = "sess-1"
+const INSTANCE_PORT = 9999
+const AUTH_HEADER = "Basic test-auth"
+const TERMINAL_TIMEOUT_MS = 3000
+
+interface CapturedRequest {
+  method: string
+  url: string
+  headers: Headers
+  body: string
+}
+
+/**
+ * Drives the real {@link BackgroundProcessManager} lifecycle (spawn a
+ * fast-exiting command with notify enabled) against a mocked transport, so the
+ * migrated `sendCompletionPrompt` path — factory + SDK client + `fetch` — is
+ * exercised end to end without touching production wiring.
+ *
+ * The workspace temp directory is intentionally left in place (under
+ * `os.tmpdir()`, OS-reaped): removing it from the test races the manager's
+ * asynchronous finalization writes, which intermittently fail with ENOENT.
+ */
+async function runCompletionPrompt(
+  fetchImpl: (input: Request, init: RequestInit | undefined) => Promise<Response>,
+): Promise<{ requests: CapturedRequest[]; warned: boolean; directory: string }> {
+  const requests: CapturedRequest[] = []
+  const originalFetch = globalThis.fetch
+  // Captured now but swapped in only inside the try below, so a failure during
+  // setup (mkdtemp, manager construction) can't leak the mocked fetch.
+  const fetchMock = (async (input: any, init: any) => {
+    const req = input instanceof Request ? input : new Request(String(input), init)
+    requests.push({
+      method: req.method,
+      url: req.url,
+      headers: req.headers,
+      body: await req.text(),
+    })
+    return fetchImpl(input instanceof Request ? input : req, init)
+  }) as typeof fetch
+
+  let warned = false
+  const logger = {
+    warn: () => { warned = true },
+    debug: () => {},
+    trace: () => {},
+    info: () => {},
+    error: () => {},
+    fatal: () => {},
+    isLevelEnabled: () => false,
+    level: "info",
+    child: () => logger,
+  } as unknown as Logger
+
+  const workspacePath = await fs.mkdtemp(path.join(os.tmpdir(), "bp-test-"))
+  // Distinct from the workspace root so the directory-override assertion is
+  // discriminating: if `sendCompletionPrompt` stops passing `notify.directory`,
+  // the factory would fall back to `workspacePath` and the header check fails.
+  const sessionDir = path.join(workspacePath, "session-worktree")
+
+  // Resolve once the manager publishes a terminal (non-running) status update.
+  let resolveTerminal: () => void = () => {}
+  const terminal = new Promise<void>((resolve) => { resolveTerminal = resolve })
+  const eventBus = {
+    on: () => {},
+    publish: (event: any) => {
+      if (event?.type === "instance.event") {
+        const type = event?.event?.type
+        const status = event?.event?.properties?.process?.status
+        if (type === "background.process.removed" || (status && status !== "running")) resolveTerminal()
+      }
+      return true
+    },
+  } as unknown as EventBus
+
+  const workspaceManager = {
+    get: () => ({ path: workspacePath }),
+    getInstancePort: () => INSTANCE_PORT,
+    getInstanceAuthorizationHeader: () => AUTH_HEADER,
+  } as unknown as WorkspaceManager
+
+  const manager = new BackgroundProcessManager({ workspaceManager, eventBus, logger })
+
+  try {
+    globalThis.fetch = fetchMock
+    await manager.start(WORKSPACE_ID, "test-proc", "true", {
+      notify: true,
+      notification: { sessionID: SESSION_ID, directory: sessionDir },
+    })
+    // The terminal status update is published at the very end of finalize, so
+    // resolving on it is a deterministic completion signal. Fail loudly rather
+    // than racing a silent timeout that could mask a hang.
+    let timeoutHandle: NodeJS.Timeout | undefined
+    const reachedTerminal = await Promise.race([
+      terminal.then(() => true),
+      new Promise<boolean>((resolve) => {
+        timeoutHandle = setTimeout(() => resolve(false), TERMINAL_TIMEOUT_MS)
+      }),
+    ])
+    if (timeoutHandle) clearTimeout(timeoutHandle)
+    if (!reachedTerminal) {
+      throw new Error("background process did not reach a terminal state in time")
+    }
+  } finally {
+    globalThis.fetch = originalFetch
+  }
+
+  return { requests, warned, directory: sessionDir }
+}
+
+describe("BackgroundProcessManager.sendCompletionPrompt", () => {
+  it("posts the synthetic completion prompt to the instance via the SDK route", async () => {
+    const { requests, directory } = await runCompletionPrompt(async () =>
+      new Response("{}", { status: 200, headers: { "content-type": "application/json" } }),
+    )
+
+    const promptCall = requests.find((r) => r.url.includes("/prompt_async"))
+    assert.ok(promptCall, "expected a prompt_async request")
+    assert.equal(promptCall.method, "POST")
+    assert.equal(
+      promptCall.url,
+      `http://127.0.0.1:${INSTANCE_PORT}/session/${SESSION_ID}/prompt_async`,
+    )
+    assert.equal(promptCall.headers.get("authorization"), AUTH_HEADER)
+    // The prompt is scoped to the session's directory (a POST keeps the
+    // directory as a header — the SDK only rewrites header→query for GET/HEAD).
+    assert.equal(promptCall.headers.get("x-opencode-directory"), encodeURIComponent(directory))
+
+    const body = JSON.parse(promptCall.body)
+    assert.equal(body.parts.length, 1)
+    assert.equal(body.parts[0].type, "text")
+    assert.equal(body.parts[0].synthetic, true)
+    assert.match(body.parts[0].text, /test-proc/)
+  })
+
+  it("swallows a failed prompt and logs it without aborting finalization", async () => {
+    const { warned } = await runCompletionPrompt(async () =>
+      new Response("boom", { status: 500 }),
+    )
+    assert.equal(warned, true)
+  })
+})

packages/server/src/background-processes/manager.ts

@@ -4,6 +4,7 @@ import path from "path"
 import { randomBytes } from "crypto"
 import type { EventBus } from "../events/bus"
 import type { WorkspaceManager } from "../workspaces/manager"
+import { createInstanceClient } from "../workspaces/instance-client"
 import type { Logger } from "../logger"
 import type { BackgroundProcess, BackgroundProcessStatus, BackgroundProcessTerminalReason } from "../api-types"
 
@@ -625,43 +626,26 @@ export class BackgroundProcessManager {
     const notify = record.notify
     if (!notify || !record.terminalReason) return
 
-    if (!this.deps.workspaceManager.get(workspaceId)) {
-      throw new Error("Workspace not found")
-    }
-
-    const port = this.deps.workspaceManager.getInstancePort(workspaceId)
-    if (!port) {
+    const client = createInstanceClient(this.deps.workspaceManager, workspaceId, {
+      directory: notify.directory,
+    })
+    if (!client) {
       throw new Error("Workspace instance is not ready")
     }
 
-    const targetUrl = `http://127.0.0.1:${port}/session/${encodeURIComponent(notify.sessionID)}/prompt_async`
-    const headers: Record<string, string> = {
-      "content-type": "application/json",
-    }
-
-    const authorization = this.deps.workspaceManager.getInstanceAuthorizationHeader(workspaceId)
-    if (authorization) {
-      headers.authorization = authorization
-    }
-
-    const response = await fetch(targetUrl, {
-      method: "POST",
-      headers,
-      body: JSON.stringify({
+    await client.session.promptAsync(
+      {
+        sessionID: notify.sessionID,
         parts: [
           {
             type: "text",
             text: this.buildSyntheticCompletionPrompt(record),
             synthetic: true,
           },
         ],
-      }),
-    })
-
-    if (!response.ok) {
-      const message = await response.text().catch(() => "")
-      throw new Error(message || `Prompt request failed with ${response.status}`)
-    }
+      },
+      { throwOnError: true },
+    )
   }
 
   private buildCompletionPrompt(record: PersistedBackgroundProcess): string {

packages/server/src/workspaces/instance-client.test.ts

@@ -0,0 +1,175 @@
+import assert from "node:assert/strict"
+import { describe, it } from "node:test"
+
+import { createInstanceClient } from "./instance-client"
+import type { WorkspaceManager } from "./manager"
+
+/**
+ * Minimal stand-in for the parts of {@link WorkspaceManager} the factory reads.
+ * The factory only touches three members, so a structural stub is enough and
+ * keeps the test free of the full manager's heavy dependencies.
+ */
+interface StubWorkspaceManager {
+  getInstancePort: (id: string) => number | undefined
+  getInstanceAuthorizationHeader: (id: string) => string | undefined
+  get: (id: string) => { path: string } | undefined
+}
+
+function makeManager(overrides: Partial<StubWorkspaceManager> = {}): StubWorkspaceManager {
+  return {
+    getInstancePort: overrides.getInstancePort ?? (() => undefined),
+    getInstanceAuthorizationHeader: overrides.getInstanceAuthorizationHeader ?? (() => undefined),
+    get: overrides.get ?? (() => undefined),
+  }
+}
+
+interface CapturedRequest {
+  url: string
+  headers: Headers
+}
+
+/**
+ * Installs a global `fetch` stub that records every outgoing request and
+ * answers a minimal healthy JSON body. Returns the capture buffer and a
+ * restore function. The stub tolerates both `fetch(url, init)` and
+ * `fetch(Request)` invocation styles so it is independent of the SDK's
+ * internal call convention.
+ */
+function installRecordingFetch(): { requests: CapturedRequest[]; restore: () => void } {
+  const requests: CapturedRequest[] = []
+  const original = globalThis.fetch
+
+  globalThis.fetch = (async (input: any, init: any) => {
+    if (input instanceof Request) {
+      requests.push({ url: input.url, headers: new Headers(init?.headers ?? input.headers) })
+    } else {
+      requests.push({ url: String(input), headers: new Headers(init?.headers) })
+    }
+    return new Response(JSON.stringify({ healthy: true }), {
+      status: 200,
+      headers: { "content-type": "application/json" },
+    })
+  }) as typeof fetch
+
+  return { requests, restore: () => { globalThis.fetch = original } }
+}
+
+describe("createInstanceClient", () => {
+  it("returns null when the instance has no open port", () => {
+    const manager = makeManager({ getInstancePort: () => undefined })
+    assert.equal(createInstanceClient(manager as unknown as WorkspaceManager, "ws-1"), null)
+  })
+
+  it("targets the loopback host and port on outgoing requests", async () => {
+    const { requests, restore } = installRecordingFetch()
+    try {
+      const manager = makeManager({ getInstancePort: () => 4321, get: () => ({ path: "/repo" }) })
+      const client = createInstanceClient(manager as unknown as WorkspaceManager, "ws-1")
+      assert.ok(client, "expected a client when the instance has a port")
+
+      await client!.global.health()
+      const parsed = new URL(requests[0].url)
+      assert.equal(parsed.hostname, "127.0.0.1")
+      assert.equal(parsed.port, "4321")
+    } finally {
+      restore()
+    }
+  })
+
+  it("attaches the authorization header when one is configured", async () => {
+    const { requests, restore } = installRecordingFetch()
+    try {
+      const manager = makeManager({
+        getInstancePort: () => 4321,
+        getInstanceAuthorizationHeader: () => "Basic abc",
+        get: () => ({ path: "/repo" }),
+      })
+      const client = createInstanceClient(manager as unknown as WorkspaceManager, "ws-1")
+
+      await client!.global.health()
+      assert.equal(requests[0].headers.get("authorization"), "Basic abc")
+    } finally {
+      restore()
+    }
+  })
+
+  it("omits the authorization header when none is configured", async () => {
+    const { requests, restore } = installRecordingFetch()
+    try {
+      const manager = makeManager({ getInstancePort: () => 4321, get: () => ({ path: "/repo" }) })
+      const client = createInstanceClient(manager as unknown as WorkspaceManager, "ws-1")
+
+      await client!.global.health()
+      assert.equal(requests[0].headers.get("authorization"), null)
+    } finally {
+      restore()
+    }
+  })
+
+  it("scopes requests to the workspace directory", async () => {
+    const { requests, restore } = installRecordingFetch()
+    try {
+      const manager = makeManager({ getInstancePort: () => 4321, get: () => ({ path: "/repo" }) })
+      const client = createInstanceClient(manager as unknown as WorkspaceManager, "ws-1")
+
+      await client!.global.health()
+      // GET requests carry directory as a query parameter (see SDK rewrite).
+      assert.equal(new URL(requests[0].url).searchParams.get("directory"), "/repo")
+    } finally {
+      restore()
+    }
+  })
+
+  it("does not scope requests when the workspace has no path", async () => {
+    const { requests, restore } = installRecordingFetch()
+    try {
+      const manager = makeManager({ getInstancePort: () => 4321, get: () => undefined })
+      const client = createInstanceClient(manager as unknown as WorkspaceManager, "ws-1")
+
+      await client!.global.health()
+      assert.equal(new URL(requests[0].url).searchParams.get("directory"), null)
+    } finally {
+      restore()
+    }
+  })
+
+  it("honours an explicit directory override over the workspace root", async () => {
+    const { requests, restore } = installRecordingFetch()
+    try {
+      const manager = makeManager({ getInstancePort: () => 4321, get: () => ({ path: "/workspace-root" }) })
+      const client = createInstanceClient(manager as unknown as WorkspaceManager, "ws-1", {
+        directory: "/explicit/session-dir",
+      })
+
+      await client!.global.health()
+      assert.equal(new URL(requests[0].url).searchParams.get("directory"), "/explicit/session-dir")
+    } finally {
+      restore()
+    }
+  })
+
+  it("applies the loopback timeout and aborts a stuck instance", async () => {
+    const original = globalThis.fetch
+    // Never resolves on its own; only settles when the passed signal aborts,
+    // mirroring how a real fetch honours an AbortSignal. Without the factory
+    // timeout this call would hang forever and time the test out.
+    globalThis.fetch = (async (_input: any, init: any) => {
+      return new Promise((_resolve, reject) => {
+        const signal = (init as RequestInit | undefined)?.signal
+        if (!signal) return
+        if (signal.aborted) reject((signal as AbortSignal).reason ?? new Error("aborted"))
+        else signal.addEventListener("abort", () => reject((signal as AbortSignal).reason ?? new Error("aborted")))
+      })
+    }) as typeof fetch
+    try {
+      const manager = makeManager({ getInstancePort: () => 4321, get: () => ({ path: "/repo" }) })
+      const client = createInstanceClient(manager as unknown as WorkspaceManager, "ws-1", { timeoutMs: 10 })
+
+      // SDK methods resolve with { error } rather than throwing by default.
+      const result = await client!.global.health()
+      assert.ok(result.error, "expected the stuck-instance call to surface an error")
+    } finally {
+      globalThis.fetch = original
+    }
+  })
+})