feat(core): add persistent terminal groups

packages/core/src/persistent-pty.ts

@@ -0,0 +1 @@
+export { Group } from "./persistent-pty/group.js"

packages/core/src/persistent-pty/group.ts

@@ -0,0 +1,101 @@
+export * as Group from "./group.js"
+
+import { Group } from "@opencode-ai/schema/group"
+import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
+import { Context, Effect, Layer, Schema } from "effect"
+import { Bus } from "../bus.js"
+import { KeyedMutex } from "../effect/keyed-mutex.js"
+import { KV } from "../kv.js"
+
+export const ID = Group.ID
+export type ID = Group.ID
+export const Item = Group.Item
+export type Item = Group.Item
+export const Info = Group.Info
+export type Info = Group.Info
+export const Event = Group.Event
+
+export interface Interface {
+  readonly get: (id: ID) => Effect.Effect<Info | undefined>
+  readonly create: (id: ID, items?: ReadonlyArray<Item>) => Effect.Effect<Info>
+  readonly addItem: (id: ID, item: Item) => Effect.Effect<Info | undefined>
+  readonly removeItem: (id: ID, item: Item) => Effect.Effect<Info | undefined>
+  readonly remove: (id: ID) => Effect.Effect<void>
+}
+
+export class Service extends Context.Service<Service, Interface>()("@opencode/Group") {}
+
+const layer = Layer.effect(
+  Service,
+  Effect.gen(function* () {
+    const kv = yield* KV.Service
+    const bus = yield* Bus.Service
+    const locks = KeyedMutex.makeUnsafe<ID>()
+
+    const get = Effect.fn("Group.get")(function* (id: ID) {
+      const value = yield* kv.get(`group:v1:${id}`)
+      return Schema.is(Info)(value) ? value : undefined
+    })
+
+    return Service.of({
+      get,
+      create: Effect.fn("Group.create")(function* (id, items = []) {
+        return yield* locks.withLock(id)(
+          Effect.gen(function* () {
+            const existing = yield* get(id)
+            if (existing) return existing
+            const group = Info.make({ id, items: unique(items) })
+            yield* kv.set(`group:v1:${id}`, group)
+            return group
+          }),
+        )
+      }),
+      addItem: Effect.fn("Group.addItem")(function* (id, item) {
+        return yield* locks.withLock(id)(
+          Effect.gen(function* () {
+            const group = yield* get(id)
+            if (!group || group.items.some((current) => same(current, item))) return group
+            const updated = Info.make({ id, items: group.items.concat(item) })
+            yield* kv.set(`group:v1:${id}`, updated)
+            yield* bus.publish(Event.ItemAdded, { groupID: id, item })
+            return updated
+          }),
+        )
+      }),
+      removeItem: Effect.fn("Group.removeItem")(function* (id, item) {
+        return yield* locks.withLock(id)(
+          Effect.gen(function* () {
+            const group = yield* get(id)
+            if (!group || !group.items.some((current) => same(current, item))) return group
+            const updated = Info.make({ id, items: group.items.filter((current) => !same(current, item)) })
+            yield* kv.set(`group:v1:${id}`, updated)
+            yield* bus.publish(Event.ItemRemoved, { groupID: id, item })
+            return updated
+          }),
+        )
+      }),
+      remove: Effect.fn("Group.remove")(function* (id) {
+        yield* locks.withLock(id)(
+          Effect.gen(function* () {
+            const group = yield* get(id)
+            yield* kv.remove(`group:v1:${id}`)
+            if (!group) return
+            yield* Effect.forEach(group.items, (item) => bus.publish(Event.ItemRemoved, { groupID: id, item }), {
+              discard: true,
+            })
+          }),
+        )
+      }),
+    })
+  }),
+)
+
+function same(left: Item, right: Item) {
+  return left.type === right.type && left.id === right.id
+}
+
+function unique(items: ReadonlyArray<Item>) {
+  return items.filter((item, index) => items.findIndex((current) => same(current, item)) === index)
+}
+
+export const node = makeGlobalNode({ service: Service, layer, deps: [KV.node, Bus.node] })

packages/core/test/group.test.ts

@@ -0,0 +1,107 @@
+import { describe, expect } from "bun:test"
+import { Bus } from "@opencode-ai/core/bus"
+import { KV } from "@opencode-ai/core/kv"
+import { Group } from "@opencode-ai/core/persistent-pty"
+import { Pty } from "@opencode-ai/schema/pty"
+import { Session } from "@opencode-ai/schema/session"
+import { LayerNode } from "@opencode-ai/util/effect/layer-node"
+import { Effect, Fiber, Stream } from "effect"
+import { testEffect } from "./lib/effect"
+
+const it = testEffect(LayerNode.compile(LayerNode.group([Group.node, KV.node, Bus.node])))
+
+describe("Group", () => {
+  it.effect("persists each group in its own versioned KV entry", () =>
+    Effect.gen(function* () {
+      const groups = yield* Group.Service
+      const kv = yield* KV.Service
+      const id = Group.ID.make("grp_session_one")
+      const created = yield* groups.create(id, [
+        { type: "session", id: Session.ID.make("ses_one") },
+        { type: "terminal", id: Pty.ID.make("pty_one") },
+        { type: "terminal", id: Pty.ID.make("pty_one") },
+      ])
+
+      expect(yield* groups.get(id)).toEqual(created)
+      expect(yield* kv.get(`group:v1:${id}`)).toEqual(created)
+      expect(created.items).toHaveLength(2)
+
+      yield* groups.removeItem(id, { type: "session", id: Session.ID.make("ses_one") })
+      yield* groups.addItem(id, { type: "terminal", id: Pty.ID.make("pty_two") })
+      expect((yield* groups.get(id))?.items).toEqual([
+        { type: "terminal", id: Pty.ID.make("pty_one") },
+        { type: "terminal", id: Pty.ID.make("pty_two") },
+      ])
+
+      yield* groups.remove(id)
+      expect(yield* groups.get(id)).toBeUndefined()
+      expect(yield* kv.get(`group:v1:${id}`)).toBeUndefined()
+    }),
+  )
+
+  it.effect("creates caller-owned group IDs idempotently", () =>
+    Effect.gen(function* () {
+      const groups = yield* Group.Service
+      const id = Group.ID.make("grp_session_one")
+      const first = yield* groups.create(id, [{ type: "session", id: Session.ID.make("ses_one") }])
+      const second = yield* groups.create(id, [{ type: "session", id: Session.ID.make("ses_other") }])
+
+      expect(second).toEqual(first)
+      expect(yield* groups.get(id)).toEqual(first)
+    }),
+  )
+
+  it.effect("serializes concurrent mutations within one group", () =>
+    Effect.gen(function* () {
+      const groups = yield* Group.Service
+      const id = Group.ID.make("grp_session_one")
+      yield* groups.create(id)
+      yield* Effect.all(
+        Array.from({ length: 20 }, (_, index) =>
+          groups.addItem(id, { type: "terminal", id: Pty.ID.make(`pty_${index}`) }),
+        ),
+        { concurrency: "unbounded" },
+      )
+
+      expect((yield* groups.get(id))?.items).toHaveLength(20)
+    }),
+  )
+
+  it.effect("publishes every removed group item", () =>
+    Effect.gen(function* () {
+      const groups = yield* Group.Service
+      const bus = yield* Bus.Service
+      const session = { type: "session" as const, id: Session.ID.make("ses_one") }
+      const terminal = { type: "terminal" as const, id: Pty.ID.make("pty_one") }
+      const group = yield* groups.create(Group.ID.make("grp_session_one"), [session, terminal])
+      const events = yield* bus
+        .subscribe(Group.Event.ItemRemoved)
+        .pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped)
+      yield* Effect.yieldNow
+
+      yield* groups.removeItem(group.id, terminal)
+      yield* groups.remove(group.id)
+
+      expect(Array.from(yield* Fiber.join(events)).map((event) => event.data)).toEqual([
+        { groupID: group.id, item: terminal },
+        { groupID: group.id, item: session },
+      ])
+    }),
+  )
+
+  it.effect("publishes every added group item", () =>
+    Effect.gen(function* () {
+      const groups = yield* Group.Service
+      const bus = yield* Bus.Service
+      const session = { type: "session" as const, id: Session.ID.make("ses_one") }
+      const terminal = { type: "terminal" as const, id: Pty.ID.make("pty_one") }
+      const group = yield* groups.create(Group.ID.make("grp_session_one"), [session])
+      const event = yield* bus.subscribe(Group.Event.ItemAdded).pipe(Stream.runHead, Effect.forkScoped)
+      yield* Effect.yieldNow
+
+      yield* groups.addItem(group.id, terminal)
+
+      expect((yield* Fiber.join(event)).valueOrUndefined?.data).toEqual({ groupID: group.id, item: terminal })
+    }),
+  )
+})

packages/schema/src/event-manifest.ts

@@ -10,6 +10,7 @@ import { Event } from "./event.js"
 import { FileSystem } from "./filesystem.js"
 import { FileSystemV1 } from "./filesystem-v1.js"
 import { Form } from "./form.js"
+import { Group } from "./group.js"
 import { InstallationEvent } from "./installation-event.js"
 import { Integration } from "./integration.js"
 import { LegacyEventV1 } from "./legacy-event.js"
@@ -56,6 +57,7 @@ const featureDefinitions = Event.inventory(
   ...Pty.Event.Definitions,
   ...Shell.Event.Definitions,
   ...Form.Event.Definitions,
+  ...Group.Event.Definitions,
   ...WebSearch.Event.Definitions,
 )
 

packages/schema/src/group.ts

@@ -0,0 +1,43 @@
+export * as Group from "./group.js"
+
+import { Schema } from "effect"
+import { ephemeral, inventory } from "./event.js"
+import { ascending } from "./identifier.js"
+import { Pty } from "./pty.js"
+import { statics } from "./schema.js"
+import { Session } from "./session.js"
+
+const IDSchema = Schema.String.check(Schema.isStartsWith("grp_")).pipe(Schema.brand("GroupID"))
+
+export const ID = IDSchema.pipe(
+  statics((schema: typeof IDSchema) => ({ create: () => schema.make("grp_" + ascending()) })),
+)
+export type ID = typeof ID.Type
+
+export const SessionItem = Schema.Struct({
+  type: Schema.tag("session"),
+  id: Session.ID,
+})
+export interface SessionItem extends Schema.Schema.Type<typeof SessionItem> {}
+
+export const TerminalItem = Schema.Struct({
+  type: Schema.tag("terminal"),
+  id: Pty.ID,
+})
+export interface TerminalItem extends Schema.Schema.Type<typeof TerminalItem> {}
+
+export const Item = Schema.Union([SessionItem, TerminalItem]).pipe(
+  Schema.toTaggedUnion("type"),
+  Schema.annotate({ identifier: "Group.Item" }),
+)
+export type Item = typeof Item.Type
+
+export const Info = Schema.Struct({
+  id: ID,
+  items: Schema.Array(Item),
+}).annotate({ identifier: "Group.Info" })
+export interface Info extends Schema.Schema.Type<typeof Info> {}
+
+const ItemAdded = ephemeral({ type: "group.item.added", schema: { groupID: ID, item: Item } })
+const ItemRemoved = ephemeral({ type: "group.item.removed", schema: { groupID: ID, item: Item } })
+export const Event = { ItemAdded, ItemRemoved, Definitions: inventory(ItemAdded, ItemRemoved) }

packages/schema/src/index.ts

@@ -6,6 +6,7 @@ export { Credential } from "./credential.js"
 export { Event } from "./event.js"
 export { FileSystem } from "./filesystem.js"
 export { Form } from "./form.js"
+export { Group } from "./group.js"
 export { Integration } from "./integration.js"
 export { LLM } from "./llm.js"
 export { Location } from "./location.js"

packages/schema/test/event-manifest.test.ts

@@ -4,6 +4,7 @@ import {
   Config,
   FileSystem,
   Form,
+  Group,
   Integration,
   Permission,
   Project,
@@ -64,6 +65,7 @@ describe("public event manifest", () => {
     expect(Integration.Event.Definitions).toEqual([Integration.Event.Updated, Integration.Event.ConnectionUpdated])
     expect(Permission.Event.Definitions).toEqual([Permission.Event.Asked, Permission.Event.Replied])
     expect(Form.Event.Definitions).toEqual([Form.Event.Created, Form.Event.Replied, Form.Event.Cancelled])
+    expect(Group.Event.Definitions).toEqual([Group.Event.ItemAdded, Group.Event.ItemRemoved])
     expect(Reference.Event.Definitions).toEqual([Reference.Event.Updated])
     expect(Plugin.Event.Definitions).toEqual([Plugin.Event.Added, Plugin.Event.Updated])
     expect(McpEvent.Definitions).toEqual([McpEvent.ToolsChanged, McpEvent.ResourcesChanged, McpEvent.StatusChanged])

packages/schema/test/group.test.ts

@@ -0,0 +1,31 @@
+import { describe, expect, test } from "bun:test"
+import { Schema } from "effect"
+import { Group } from "../src/group.js"
+import { Pty } from "../src/pty.js"
+import { Session } from "../src/session.js"
+
+describe("Group", () => {
+  test("creates branded group IDs", () => {
+    expect(Group.ID.create()).toStartWith("grp_")
+    expect(() => Schema.decodeUnknownSync(Group.ID)("ses_invalid")).toThrow()
+  })
+
+  test("preserves one ordered session and terminal item list", () => {
+    const group = Schema.decodeUnknownSync(Group.Info)({
+      id: Group.ID.create(),
+      items: [
+        { type: "session", id: Session.ID.make("ses_one") },
+        { type: "terminal", id: Pty.ID.make("pty_one") },
+        { type: "session", id: Session.ID.make("ses_two") },
+      ],
+    })
+
+    expect(group.items.map((item) => item.type)).toEqual(["session", "terminal", "session"])
+    expect(() =>
+      Schema.decodeUnknownSync(Group.Info)({
+        id: group.id,
+        items: [{ type: "other", id: "other_one" }],
+      }),
+    ).toThrow()
+  })
+})