fix(core): gate durable event persistence

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

packages/core/src/event.ts

@@ -10,6 +10,7 @@ import { Location } from "./location"
 import { makeGlobalNode } from "./effect/app-node"
 import { isDeepStrictEqual } from "node:util"
 import { Durable } from "@opencode-ai/schema/durable-event-manifest"
+import { truthy } from "./flag/flag"
 
 export const ID = Event.ID
 export type ID = import("@opencode-ai/schema/event").ID
@@ -165,6 +166,7 @@ export const allBounded = (events: Interface, capacity: number) =>
 
 export interface LayerOptions {
   readonly beforeAggregateRead?: (aggregateID: string) => Effect.Effect<void>
+  readonly persistDurableEvents?: boolean
 }
 
 export const layerWith = (options?: LayerOptions) =>
@@ -180,6 +182,7 @@ export const layerWith = (options?: LayerOptions) =>
       // TODO: Bind durable projectors to exact type+version before supporting incompatible historical payloads.
       const listeners = new Array<Subscriber>()
       const { db } = yield* Database.Service
+      const persistDurableEvents = options?.persistDurableEvents ?? truthy("OPENCODE_EXPERIMENTAL_EVENT_PERSISTENCE")
 
       const getOrCreate = (definition: Definition) =>
         Effect.gen(function* () {
@@ -333,19 +336,20 @@ export const layerWith = (options?: LayerOptions) =>
                             })
                             .run()
                             .pipe(Effect.orDie)
-                          yield* db
-                            .insert(EventTable)
-                            .values([
-                              {
-                                id: event.id,
-                                aggregate_id: aggregateID,
-                                seq,
-                                type: versionedType(definition.type, durable.version),
-                                data: encoded,
-                              },
-                            ])
-                            .run()
-                            .pipe(Effect.orDie)
+                          if (persistDurableEvents)
+                            yield* db
+                              .insert(EventTable)
+                              .values([
+                                {
+                                  id: event.id,
+                                  aggregate_id: aggregateID,
+                                  seq,
+                                  type: versionedType(definition.type, durable.version),
+                                  data: encoded,
+                                },
+                              ])
+                              .run()
+                              .pipe(Effect.orDie)
                           return { aggregateID, seq }
                         }),
                       { behavior: "immediate" },

packages/core/test/event.test.ts

@@ -82,6 +82,9 @@ const it = testEffect(
   AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, Location.node]), [[Location.node, locationLayer]]),
 )
 const itWithoutLocation = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node])))
+const itWithoutHistory = testEffect(
+  EventV2.layerWith({ persistDurableEvents: false }).pipe(Layer.provideMerge(AppNodeBuilder.build(Database.node))),
+)
 
 describe("EventV2", () => {
   it.effect("publishes events with the current location", () =>
@@ -400,6 +403,29 @@ describe("EventV2", () => {
     }),
   )
 
+  itWithoutHistory.effect("commits projections and sequences without retaining event history", () =>
+    Effect.gen(function* () {
+      const events = yield* EventV2.Service
+      const { db } = yield* Database.Service
+      const aggregateID = EventV2.ID.create()
+      const projected = new Array<number>()
+      yield* events.project(SyncMessage, (event) =>
+        Effect.sync(() => {
+          projected.push(event.durable!.seq)
+        }),
+      )
+
+      yield* events.publish(SyncMessage, { id: aggregateID, text: "first" })
+      yield* events.publish(SyncMessage, { id: aggregateID, text: "second" })
+
+      expect(projected).toEqual([0, 1])
+      expect(yield* db.select().from(EventTable).where(eq(EventTable.aggregate_id, aggregateID)).all()).toEqual([])
+      expect(
+        yield* db.select().from(EventSequenceTable).where(eq(EventSequenceTable.aggregate_id, aggregateID)).get(),
+      ).toMatchObject({ aggregate_id: aggregateID, seq: 1 })
+    }),
+  )
+
   it.effect("increments durable event seq per aggregate", () =>
     Effect.gen(function* () {
       const events = yield* EventV2.Service

packages/core/test/preload.ts

@@ -3,3 +3,4 @@ import path from "path"
 process.env.OPENCODE_DB = ":memory:"
 process.env.OPENCODE_MODELS_PATH = path.join(import.meta.dir, "plugin", "fixtures", "models-dev.json")
 process.env.OPENCODE_DISABLE_MODELS_FETCH = "true"
+process.env.OPENCODE_EXPERIMENTAL_EVENT_PERSISTENCE = "true"

packages/opencode/src/acp/service.ts

@@ -656,7 +656,7 @@ function makeUsageService(sdk: OpencodeClient) {
           sessionId: params.sessionID,
           update: {
             sessionUpdate: "usage_update",
-            used: UsageService.contextTokens(message),
+            used: message.tokens.input + message.tokens.cache.read,
             size,
             cost: { amount: UsageService.totalSessionCost(messages), currency: "USD" },
           },

packages/opencode/src/acp/usage.ts

@@ -83,10 +83,6 @@ export function messageLoaderFromSDK(sdk: SDK): MessageLoaderInterface {
 
 export const messageLoaderLayer = (sdk: SDK) => Layer.succeed(MessageLoader, messageLoaderFromSDK(sdk))
 
-export function contextTokens(message: AssistantTokenCost): number {
-  return message.tokens.input + message.tokens.cache.read + message.tokens.cache.write
-}
-
 export function buildUsage(message: AssistantTokenCost): Usage {
   const cachedReadTokens = message.tokens.cache.read
   const cachedWriteTokens = message.tokens.cache.write
@@ -211,7 +207,7 @@ const layer = Layer.effect(
             sessionId: input.sessionID,
             update: {
               sessionUpdate: "usage_update",
-              used: contextTokens(message),
+              used: message.tokens.input + message.tokens.cache.read,
               size,
               cost: { amount: totalSessionCost(messages), currency: "USD" },
             },

packages/opencode/src/server/routes/instance/httpapi/middleware/proxy.ts

@@ -97,6 +97,29 @@ export function http(
     headers.delete("content-encoding")
     headers.delete("content-length")
 
+    // An upstream 5xx from a remote workspace sandbox arrives here as an opaque
+    // status — its real cause (and log line) live only inside the sandbox. Buffer
+    // the small error body, log it locally so it shows up in the host's log, and
+    // forward it unchanged (preserving content-type so the client can still parse
+    // the structured error, e.g. its `ref`).
+    if (response.status >= 500) {
+      const body = yield* response.text.pipe(Effect.catch(() => Effect.succeed("")))
+      const contentType = response.headers["content-type"] ?? "application/json"
+      headers.delete("content-type")
+      yield* Effect.logError("workspace proxy upstream error", {
+        url: url.toString(),
+        method: request.method,
+        status: response.status,
+        body: body.slice(0, 2000),
+      })
+      return HttpServerResponse.text(body, {
+        status: response.status,
+        statusText: statusText(response),
+        headers,
+        contentType,
+      })
+    }
+
     return HttpServerResponse.stream(response.stream.pipe(Stream.catchCause(() => Stream.empty)), {
       status: response.status,
       statusText: statusText(response),

packages/opencode/src/server/shared/workspace-routing.ts

@@ -34,5 +34,12 @@ export function workspaceProxyURL(target: string | URL, requestURL: URL) {
   proxyURL.search = requestURL.search
   proxyURL.hash = requestURL.hash
   proxyURL.searchParams.delete("workspace")
+  // The `directory` param is the *host's* working directory (e.g. a Windows
+  // path like `F:\proj`). It is meaningless — and dangerous — on the remote:
+  // the sandbox would `path.resolve` it against its own cwd, producing a bogus
+  // path like `/home/daytona/workspace/repo/F:\proj` that does not exist and
+  // crashes prompt handling. Drop it so the remote falls back to its own
+  // project root. This mirrors ProxyUtil.headers stripping `x-opencode-directory`.
+  proxyURL.searchParams.delete("directory")
   return proxyURL
 }

packages/opencode/test/acp/usage.test.ts

@@ -207,7 +207,7 @@ describe("acp usage", () => {
     )
   })
 
-  it.effect("includes cache reads and writes in ACP context usage", () => {
+  it.effect("sends ACP usage_update with context size and cumulative assistant cost", () => {
     const updates: SessionNotification[] = []
     return Effect.gen(function* () {
       const usage = yield* UsageService.Service
@@ -222,7 +222,7 @@ describe("acp usage", () => {
           sessionId: "ses_1",
           update: {
             sessionUpdate: "usage_update",
-            used: 22,
+            used: 15,
             size: 128_000,
             cost: { amount: 3, currency: "USD" },
           },
@@ -239,7 +239,7 @@ describe("acp usage", () => {
                 input: 10,
                 output: 20,
                 reasoning: 0,
-                cache: { read: 5, write: 7 },
+                cache: { read: 5, write: 0 },
               },
             }),
           ]),

packages/opencode/test/preload.ts

@@ -37,6 +37,7 @@ process.env["XDG_CONFIG_HOME"] = path.join(dir, "config")
 process.env["XDG_STATE_HOME"] = path.join(dir, "state")
 process.env["OPENCODE_MODELS_PATH"] = path.join(import.meta.dir, "tool", "fixtures", "models-api.json")
 process.env["OPENCODE_EXPERIMENTAL_EVENT_SYSTEM"] = "true"
+process.env["OPENCODE_EXPERIMENTAL_EVENT_PERSISTENCE"] = "true"
 process.env["OPENCODE_EXPERIMENTAL_WORKSPACES"] = "true"
 
 // Set test home directory to isolate tests from user's actual home directory

packages/opencode/test/server/workspace-routing.test.ts

@@ -80,6 +80,13 @@ describe("workspaceProxyURL", () => {
     expect(result.searchParams.get("keep")).toBe("yes")
   })
 
+  test("strips the host directory param so the remote resolves its own root", () => {
+    const url = new URL("http://localhost/session/abc?directory=F%3A%5Cproj&keep=yes")
+    const result = workspaceProxyURL("http://remote:8080/base", url)
+    expect(result.searchParams.get("directory")).toBeNull()
+    expect(result.searchParams.get("keep")).toBe("yes")
+  })
+
   test("preserves hash from request", () => {
     const url = new URL("http://localhost/page#section")
     const result = workspaceProxyURL("http://remote:8080", url)