feat(core): add v2 manual compact, drain failure event, and v1 prompt bridge

Author: See U

MERGE.md

@@ -160,6 +160,11 @@ fork 与上游改了同一处（常见于 TUI 视觉/UX、core 加固逻辑）
 | `packages/server/src/cors.ts` (CORS origin 精确匹配) | 低 | 保留精确正则匹配（`^https?:\/\/(localhost|127\.0\.0\.1):(\d+)$`）替代 `startsWith`，防 origin 欺骗（上游若已做可取上游版本） |
 | `packages/server/src/auth.ts` (timing-safe 密码比较) | 低 | 保留 `crypto.timingSafeEqual` + 长度前置检查，防时序攻击（上游若已做可取上游版本） |
 | `packages/core/src/control-plane/workspace.sql.ts` (project_id 索引) | 低 | 保留 `index("workspace_project_idx").on(table.project_id)`，加速 workspace 按 project 查询（上游若已做可取上游版本） |
+| `packages/core/src/catalog.ts` (`available()` 接受 `api.settings.apiKey`) | 中 | 保留：V1 config 迁移把 `options.apiKey` 降为 `api.settings.apiKey` 且注册无连接的 env integration，导致可用性判定与模型解析（`session/runner/model.ts`）不一致、v2 写路径 `ModelUnavailableError`；回归测试 `packages/core/test/catalog.test.ts`（derives availability from api settings apiKey）；配套 fork 新增影子对比 harness `packages/opencode/test/server/session-shadow.test.ts`（v1/v2 写路径 parity）（上游若已做可取上游版本） |
+| `packages/opencode/src/session/prompt.ts` (v1 prompt 桥接 v2 读投影) | 中 | 保留：`prompt()` 在 `experimentalEventSystem` 下发布 `session.next.prompted` + 每个 synthetic 文本部分一条 `session.next.synthetic`，使 v1 会话对 `SessionV2.messages`（`session_message` 投影）可见；实现的是上游被跳过测试 `test/session/prompt.test.ts`「prompt emits v2 prompted and synthetic events」定义的合同（已解除 skip），并同步改写「legacy prompt ... without session.next events」为桥接后合同（上游若实现同等桥接则取上游版本并还原测试） |
+| `packages/schema/src/session-event.ts` + `packages/core/src/session/runner/llm.ts` (v2 drain 失败 live 事件 `session.next.failed`) | 中 | 保留：新增 live-only `SessionEvent.Failed`（不入 durable manifest/不投影），`SessionRunner.run` 失败（非纯中断 cause）时发布，覆盖 step 开始前的失败（如 `ModelUnavailableError`）——此前 drain 失败仅 `execution/local.ts` logError，客户端不可见；step 内失败仍由 durable `Step.Failed` 结算（会双发，与 v1 `session.error` + assistant error 双通道一致）；测试 `packages/core/test/session-runner.test.ts`（publishes a live failed event when a drain fails）；连带更新 manifest 计数测试（schema 侧原本在 HEAD 即过期失败，修正为真实计数 89/59/35）+ SDK v2 重新生成（上游 runner 若补 durable 状态事件则取上游版本并移除本事件） |
+| `packages/tui/src/context/sdk.tsx` + `packages/sdk/js/src/v2/gen` (移除残留 `/sync/*` SDK 面) | 低 | 保留：fork 早前裁剪了 server `/sync/*` 路由，但 SDK gen 与 TUI `sdk.sync.start()`（实验 flag 下静默 404）未同步；本次 SDK 重新生成后一并删除 TUI 调用点（上游 sync 面回归时随上游恢复） |
+| `packages/core/src/{session.ts,session/compaction.ts,session/history.ts,session/runner/*}` (v2 手动 compact 实现) | 中 | 保留：实现 `SessionV2.compact`（原 stub 恒 `OperationUnavailableError`），spec `specs/v2/session.md:105` 已列为 sanctioned improvement。`SessionCompaction.make` 抽出共享 `summarize` 核心（`reason: "auto"\|"manual"`），新增 `compactManually`（输出预算取 `model.route.defaults.limits?.output`，可选 `instructions` 折入 `buildPrompt`）；`SessionHistory.load` 改由新导出 `entries()`（读 epoch `baseline_seq` 只读，不触发 SystemContext 初始化）派生；`SessionRunner` 接口新增 `compact`，`SessionV2.compact` 经 `locations.get(session.location)` 路由，失败/无可压缩内容统一映射 `OperationUnavailableError`（HTTP 503 文案不变）。已知限制：手动 compact 不经 `SessionRunCoordinator` 串行化，与进行中 drain 并发时下一 turn 才生效；测试 `packages/core/test/session-runner.test.ts`（manually compacts on demand / returns false when nothing to compact）（上游实现 v2 手动 compact 后取上游版本） |
 
 ### TUI 偏离（四批 23 轮审计打磨，全域）
 

packages/core/src/catalog.ts

@@ -71,6 +71,9 @@ const layer = Layer.effect(
     const available = (provider: ProviderV2.Info, integration: Integration.Info | undefined) => {
       if (provider.disabled) return false
       if (typeof provider.request.body.apiKey === "string") return true
+      // V1 config migration lowers options.apiKey into api.settings.apiKey;
+      // model resolution accepts it (session/runner/model.ts), so must availability.
+      if (typeof provider.api.settings?.apiKey === "string") return true
       if (integration?.connections.length) return true
       return provider.integrationID === undefined && !integration
     }

packages/core/src/event.ts

@@ -215,154 +215,175 @@ export const layerWith = (options?: LayerOptions) =>
       ) {
         return Effect.gen(function* () {
           const durable = definition?.durable
-          if (durable) {
-            const aggregateID = (event.data as Record<string, unknown>)[durable.aggregate]
-            if (typeof aggregateID !== "string") {
-              yield* Effect.die(
-                new InvalidDurableEventError({
-                  type: event.type,
-                  message: `Expected string aggregate field ${durable.aggregate}`,
-                }),
-              )
-            } else {
-              if (input && input.aggregateID !== aggregateID) {
-                yield* Effect.die(
-                  new InvalidDurableEventError({
-                    type: event.type,
-                    message: `Aggregate mismatch: expected ${input.aggregateID}, got ${aggregateID}`,
-                  }),
+          if (!durable) return
+          const aggregateID = (event.data as Record<string, unknown>)[durable.aggregate]
+          if (typeof aggregateID !== "string")
+            return yield* Effect.die(
+              new InvalidDurableEventError({
+                type: event.type,
+                message: `Expected string aggregate field ${durable.aggregate}`,
+              }),
+            )
+          if (input && input.aggregateID !== aggregateID)
+            return yield* Effect.die(
+              new InvalidDurableEventError({
+                type: event.type,
+                message: `Aggregate mismatch: expected ${input.aggregateID}, got ${aggregateID}`,
+              }),
+            )
+          const list = projectors.get(event.type) ?? []
+          return yield* Effect.uninterruptible(
+            Effect.gen(function* () {
+              const committed = yield* db
+                .transaction(() => commitTransaction(definition, durable, event, aggregateID, list, input, commit), {
+                  behavior: "immediate",
+                })
+                .pipe(Effect.orDie)
+              if (committed) {
+                yield* Effect.forEach(
+                  pubsub.durable.get(committed.aggregateID) ?? [],
+                  (wake) => PubSub.publish(wake, undefined),
+                  { discard: true },
                 )
               }
-              const list = projectors.get(event.type) ?? []
-              return yield* Effect.uninterruptible(
-                Effect.gen(function* () {
-                  const committed = yield* db
-                    .transaction(
-                      () =>
-                        Effect.gen(function* () {
-                          const row = yield* db
-                            .select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id })
-                            .from(EventSequenceTable)
-                            .where(eq(EventSequenceTable.aggregate_id, aggregateID))
-                            .get()
-                            .pipe(Effect.orDie)
-                          const latest = row?.seq ?? -1
-                          const encoded = Schema.encodeUnknownSync(definition.data)(event.data) as Record<
-                            string,
-                            unknown
-                          >
-                          if (input?.strictOwner && row?.ownerID && row.ownerID !== input.ownerID) {
-                            yield* Effect.die(
-                              new InvalidDurableEventError({
-                                type: event.type,
-                                message: `Replay owner mismatch for aggregate ${aggregateID}: expected ${row.ownerID}, got ${input.ownerID ?? "none"}`,
-                              }),
-                            )
-                          }
-                          if (input && input.seq <= latest) {
-                            const stored = yield* db
-                              .select()
-                              .from(EventTable)
-                              .where(and(eq(EventTable.aggregate_id, aggregateID), eq(EventTable.seq, input.seq)))
-                              .get()
-                              .pipe(Effect.orDie)
-                            if (
-                              stored?.id === event.id &&
-                              stored.type === versionedType(definition.type, durable.version) &&
-                              isDeepStrictEqual(stored.data, encoded)
-                            ) {
-                              if (input.ownerID && row?.ownerID == null) {
-                                yield* db
-                                  .update(EventSequenceTable)
-                                  .set({ owner_id: input.ownerID })
-                                  .where(eq(EventSequenceTable.aggregate_id, aggregateID))
-                                  .run()
-                                  .pipe(Effect.orDie)
-                              }
-                              return
-                            }
-                            yield* Effect.die(
-                              new InvalidDurableEventError({
-                                type: event.type,
-                                message: `Replay diverged at aggregate ${aggregateID} sequence ${input.seq}`,
-                              }),
-                            )
-                          }
-                          if (input && row?.ownerID && row.ownerID !== input.ownerID) {
-                            return
-                          }
-                          const seq = input?.seq ?? latest + 1
-                          if (input && seq !== latest + 1) {
-                            yield* Effect.die(
-                              new InvalidDurableEventError({
-                                type: event.type,
-                                message: `Sequence mismatch for aggregate ${aggregateID}: expected ${latest + 1}, got ${seq}`,
-                              }),
-                            )
-                          }
-                          const stored = yield* db
-                            .select({ aggregateID: EventTable.aggregate_id, seq: EventTable.seq })
-                            .from(EventTable)
-                            .where(eq(EventTable.id, event.id))
-                            .get()
-                            .pipe(Effect.orDie)
-                          if (stored)
-                            yield* Effect.die(
-                              new InvalidDurableEventError({
-                                type: event.type,
-                                message: `Event ${event.id} already exists at aggregate ${stored.aggregateID} sequence ${stored.seq}`,
-                              }),
-                            )
-                          const committed = {
-                            ...event,
-                            durable: { aggregateID, seq, version: durable.version },
-                          } as Payload
-                          for (const projector of list) {
-                            yield* projector(committed)
-                          }
-                          if (commit) yield* commit(seq)
-                          yield* db
-                            .insert(EventSequenceTable)
-                            .values([{ aggregate_id: aggregateID, seq, owner_id: input?.ownerID }])
-                            .onConflictDoUpdate({
-                              target: EventSequenceTable.aggregate_id,
-                              set: {
-                                seq,
-                                ...(input?.ownerID && row?.ownerID == null ? { owner_id: input.ownerID } : {}),
-                              },
-                            })
-                            .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)
-                          return { aggregateID, seq }
-                        }),
-                      { behavior: "immediate" },
-                    )
-                    .pipe(Effect.orDie)
-                  if (committed) {
-                    yield* Effect.forEach(
-                      pubsub.durable.get(committed.aggregateID) ?? [],
-                      (wake) => PubSub.publish(wake, undefined),
-                      { discard: true },
-                    )
-                  }
-                  return committed
-                }),
-              )
+              return committed
+            }),
+          )
+        })
+      }
+
+      function commitTransaction(
+        definition: Definition,
+        durable: NonNullable<Definition["durable"]>,
+        event: Payload,
+        aggregateID: string,
+        list: Subscriber[],
+        input?: {
+          readonly seq: number
+          readonly aggregateID: string
+          readonly ownerID?: string
+          readonly strictOwner?: boolean
+        },
+        commit?: (seq: number) => Effect.Effect<void>,
+      ) {
+        return Effect.gen(function* () {
+          const row = yield* db
+            .select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id })
+            .from(EventSequenceTable)
+            .where(eq(EventSequenceTable.aggregate_id, aggregateID))
+            .get()
+            .pipe(Effect.orDie)
+          const latest = row?.seq ?? -1
+          const encoded = Schema.encodeUnknownSync(definition.data)(event.data) as Record<string, unknown>
+          if (input?.strictOwner && row?.ownerID && row.ownerID !== input.ownerID) {
+            yield* Effect.die(
+              new InvalidDurableEventError({
+                type: event.type,
+                message: `Replay owner mismatch for aggregate ${aggregateID}: expected ${row.ownerID}, got ${input.ownerID ?? "none"}`,
+              }),
+            )
+          }
+          if (input && input.seq <= latest)
+            return yield* reconcileReplay(definition, durable, event, aggregateID, input, encoded, row?.ownerID)
+          if (input && row?.ownerID && row.ownerID !== input.ownerID) {
+            return
+          }
+          const seq = input?.seq ?? latest + 1
+          if (input && seq !== latest + 1) {
+            yield* Effect.die(
+              new InvalidDurableEventError({
+                type: event.type,
+                message: `Sequence mismatch for aggregate ${aggregateID}: expected ${latest + 1}, got ${seq}`,
+              }),
+            )
+          }
+          const stored = yield* db
+            .select({ aggregateID: EventTable.aggregate_id, seq: EventTable.seq })
+            .from(EventTable)
+            .where(eq(EventTable.id, event.id))
+            .get()
+            .pipe(Effect.orDie)
+          if (stored)
+            yield* Effect.die(
+              new InvalidDurableEventError({
+                type: event.type,
+                message: `Event ${event.id} already exists at aggregate ${stored.aggregateID} sequence ${stored.seq}`,
+              }),
+            )
+          const committed = {
+            ...event,
+            durable: { aggregateID, seq, version: durable.version },
+          } as Payload
+          for (const projector of list) {
+            yield* projector(committed)
+          }
+          if (commit) yield* commit(seq)
+          yield* db
+            .insert(EventSequenceTable)
+            .values([{ aggregate_id: aggregateID, seq, owner_id: input?.ownerID }])
+            .onConflictDoUpdate({
+              target: EventSequenceTable.aggregate_id,
+              set: {
+                seq,
+                ...(input?.ownerID && row?.ownerID == null ? { owner_id: input.ownerID } : {}),
+              },
+            })
+            .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)
+          return { aggregateID, seq }
+        })
+      }
+
+      function reconcileReplay(
+        definition: Definition,
+        durable: NonNullable<Definition["durable"]>,
+        event: Payload,
+        aggregateID: string,
+        input: { readonly seq: number; readonly ownerID?: string },
+        encoded: Record<string, unknown>,
+        ownerID: string | null | undefined,
+      ) {
+        return Effect.gen(function* () {
+          const stored = yield* db
+            .select()
+            .from(EventTable)
+            .where(and(eq(EventTable.aggregate_id, aggregateID), eq(EventTable.seq, input.seq)))
+            .get()
+            .pipe(Effect.orDie)
+          if (
+            stored?.id === event.id &&
+            stored.type === versionedType(definition.type, durable.version) &&
+            isDeepStrictEqual(stored.data, encoded)
+          ) {
+            if (input.ownerID && ownerID == null) {
+              yield* db
+                .update(EventSequenceTable)
+                .set({ owner_id: input.ownerID })
+                .where(eq(EventSequenceTable.aggregate_id, aggregateID))
+                .run()
+                .pipe(Effect.orDie)
             }
+            return
           }
+          yield* Effect.die(
+            new InvalidDurableEventError({
+              type: event.type,
+              message: `Replay diverged at aggregate ${aggregateID} sequence ${input.seq}`,
+            }),
+          )
         })
       }
 

packages/core/src/session.ts

@@ -415,8 +415,15 @@ const layer = Layer.effect(
         })
       }),
       compact: Effect.fn("V2Session.compact")(function* (input) {
-        yield* result.get(input.sessionID)
-        return yield* new OperationUnavailableError({ operation: "compact" })
+        const session = yield* result.get(input.sessionID)
+        const compacted = yield* Effect.gen(function* () {
+          const runner = yield* SessionRunner.Service
+          return yield* runner.compact({ sessionID: session.id, instructions: input.prompt?.text || undefined })
+        }).pipe(
+          Effect.provide(locations.get(session.location)),
+          Effect.catch(() => Effect.succeed(false)),
+        )
+        if (!compacted) return yield* new OperationUnavailableError({ operation: "compact" })
       }),
       wait: Effect.fn("V2Session.wait")(function* (sessionID) {
         yield* result.get(sessionID)