Skip to content

Commit 46b2c54

Browse files
jlongsterjlongster
committed
feat(core): add persistent terminal groups
1 parent a02b0a4 commit 46b2c54

8 files changed

Lines changed: 288 additions & 0 deletions

File tree

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
export { Group } from "./persistent-pty/group.js"

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

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
export * as Group from "./group.js"
2+
3+
import { Group } from "@opencode-ai/schema/group"
4+
import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
5+
import { Context, Effect, Layer, Schema } from "effect"
6+
import { Bus } from "../bus.js"
7+
import { KeyedMutex } from "../effect/keyed-mutex.js"
8+
import { KV } from "../kv.js"
9+
10+
export const ID = Group.ID
11+
export type ID = Group.ID
12+
export const Item = Group.Item
13+
export type Item = Group.Item
14+
export const Info = Group.Info
15+
export type Info = Group.Info
16+
export const Event = Group.Event
17+
18+
export interface Interface {
19+
readonly get: (id: ID) => Effect.Effect<Info | undefined>
20+
readonly create: (id: ID, items?: ReadonlyArray<Item>) => Effect.Effect<Info>
21+
readonly addItem: (id: ID, item: Item) => Effect.Effect<Info | undefined>
22+
readonly removeItem: (id: ID, item: Item) => Effect.Effect<Info | undefined>
23+
readonly remove: (id: ID) => Effect.Effect<void>
24+
}
25+
26+
export class Service extends Context.Service<Service, Interface>()("@opencode/Group") {}
27+
28+
const layer = Layer.effect(
29+
Service,
30+
Effect.gen(function* () {
31+
const kv = yield* KV.Service
32+
const bus = yield* Bus.Service
33+
const locks = KeyedMutex.makeUnsafe<ID>()
34+
35+
const get = Effect.fn("Group.get")(function* (id: ID) {
36+
const value = yield* kv.get(`group:v1:${id}`)
37+
return Schema.is(Info)(value) ? value : undefined
38+
})
39+
40+
return Service.of({
41+
get,
42+
create: Effect.fn("Group.create")(function* (id, items = []) {
43+
return yield* locks.withLock(id)(
44+
Effect.gen(function* () {
45+
const existing = yield* get(id)
46+
if (existing) return existing
47+
const group = Info.make({ id, items: unique(items) })
48+
yield* kv.set(`group:v1:${id}`, group)
49+
return group
50+
}),
51+
)
52+
}),
53+
addItem: Effect.fn("Group.addItem")(function* (id, item) {
54+
return yield* locks.withLock(id)(
55+
Effect.gen(function* () {
56+
const group = yield* get(id)
57+
if (!group || group.items.some((current) => same(current, item))) return group
58+
const updated = Info.make({ id, items: group.items.concat(item) })
59+
yield* kv.set(`group:v1:${id}`, updated)
60+
yield* bus.publish(Event.ItemAdded, { groupID: id, item })
61+
return updated
62+
}),
63+
)
64+
}),
65+
removeItem: Effect.fn("Group.removeItem")(function* (id, item) {
66+
return yield* locks.withLock(id)(
67+
Effect.gen(function* () {
68+
const group = yield* get(id)
69+
if (!group || !group.items.some((current) => same(current, item))) return group
70+
const updated = Info.make({ id, items: group.items.filter((current) => !same(current, item)) })
71+
yield* kv.set(`group:v1:${id}`, updated)
72+
yield* bus.publish(Event.ItemRemoved, { groupID: id, item })
73+
return updated
74+
}),
75+
)
76+
}),
77+
remove: Effect.fn("Group.remove")(function* (id) {
78+
yield* locks.withLock(id)(
79+
Effect.gen(function* () {
80+
const group = yield* get(id)
81+
yield* kv.remove(`group:v1:${id}`)
82+
if (!group) return
83+
yield* Effect.forEach(group.items, (item) => bus.publish(Event.ItemRemoved, { groupID: id, item }), {
84+
discard: true,
85+
})
86+
}),
87+
)
88+
}),
89+
})
90+
}),
91+
)
92+
93+
function same(left: Item, right: Item) {
94+
return left.type === right.type && left.id === right.id
95+
}
96+
97+
function unique(items: ReadonlyArray<Item>) {
98+
return items.filter((item, index) => items.findIndex((current) => same(current, item)) === index)
99+
}
100+
101+
export const node = makeGlobalNode({ service: Service, layer, deps: [KV.node, Bus.node] })

‎packages/core/test/group.test.ts‎

Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,107 @@
1+
import { describe, expect } from "bun:test"
2+
import { Bus } from "@opencode-ai/core/bus"
3+
import { KV } from "@opencode-ai/core/kv"
4+
import { Group } from "@opencode-ai/core/persistent-pty"
5+
import { Pty } from "@opencode-ai/schema/pty"
6+
import { Session } from "@opencode-ai/schema/session"
7+
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
8+
import { Effect, Fiber, Stream } from "effect"
9+
import { testEffect } from "./lib/effect"
10+
11+
const it = testEffect(LayerNode.compile(LayerNode.group([Group.node, KV.node, Bus.node])))
12+
13+
describe("Group", () => {
14+
it.effect("persists each group in its own versioned KV entry", () =>
15+
Effect.gen(function* () {
16+
const groups = yield* Group.Service
17+
const kv = yield* KV.Service
18+
const id = Group.ID.make("grp_session_one")
19+
const created = yield* groups.create(id, [
20+
{ type: "session", id: Session.ID.make("ses_one") },
21+
{ type: "terminal", id: Pty.ID.make("pty_one") },
22+
{ type: "terminal", id: Pty.ID.make("pty_one") },
23+
])
24+
25+
expect(yield* groups.get(id)).toEqual(created)
26+
expect(yield* kv.get(`group:v1:${id}`)).toEqual(created)
27+
expect(created.items).toHaveLength(2)
28+
29+
yield* groups.removeItem(id, { type: "session", id: Session.ID.make("ses_one") })
30+
yield* groups.addItem(id, { type: "terminal", id: Pty.ID.make("pty_two") })
31+
expect((yield* groups.get(id))?.items).toEqual([
32+
{ type: "terminal", id: Pty.ID.make("pty_one") },
33+
{ type: "terminal", id: Pty.ID.make("pty_two") },
34+
])
35+
36+
yield* groups.remove(id)
37+
expect(yield* groups.get(id)).toBeUndefined()
38+
expect(yield* kv.get(`group:v1:${id}`)).toBeUndefined()
39+
}),
40+
)
41+
42+
it.effect("creates caller-owned group IDs idempotently", () =>
43+
Effect.gen(function* () {
44+
const groups = yield* Group.Service
45+
const id = Group.ID.make("grp_session_one")
46+
const first = yield* groups.create(id, [{ type: "session", id: Session.ID.make("ses_one") }])
47+
const second = yield* groups.create(id, [{ type: "session", id: Session.ID.make("ses_other") }])
48+
49+
expect(second).toEqual(first)
50+
expect(yield* groups.get(id)).toEqual(first)
51+
}),
52+
)
53+
54+
it.effect("serializes concurrent mutations within one group", () =>
55+
Effect.gen(function* () {
56+
const groups = yield* Group.Service
57+
const id = Group.ID.make("grp_session_one")
58+
yield* groups.create(id)
59+
yield* Effect.all(
60+
Array.from({ length: 20 }, (_, index) =>
61+
groups.addItem(id, { type: "terminal", id: Pty.ID.make(`pty_${index}`) }),
62+
),
63+
{ concurrency: "unbounded" },
64+
)
65+
66+
expect((yield* groups.get(id))?.items).toHaveLength(20)
67+
}),
68+
)
69+
70+
it.effect("publishes every removed group item", () =>
71+
Effect.gen(function* () {
72+
const groups = yield* Group.Service
73+
const bus = yield* Bus.Service
74+
const session = { type: "session" as const, id: Session.ID.make("ses_one") }
75+
const terminal = { type: "terminal" as const, id: Pty.ID.make("pty_one") }
76+
const group = yield* groups.create(Group.ID.make("grp_session_one"), [session, terminal])
77+
const events = yield* bus
78+
.subscribe(Group.Event.ItemRemoved)
79+
.pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped)
80+
yield* Effect.yieldNow
81+
82+
yield* groups.removeItem(group.id, terminal)
83+
yield* groups.remove(group.id)
84+
85+
expect(Array.from(yield* Fiber.join(events)).map((event) => event.data)).toEqual([
86+
{ groupID: group.id, item: terminal },
87+
{ groupID: group.id, item: session },
88+
])
89+
}),
90+
)
91+
92+
it.effect("publishes every added group item", () =>
93+
Effect.gen(function* () {
94+
const groups = yield* Group.Service
95+
const bus = yield* Bus.Service
96+
const session = { type: "session" as const, id: Session.ID.make("ses_one") }
97+
const terminal = { type: "terminal" as const, id: Pty.ID.make("pty_one") }
98+
const group = yield* groups.create(Group.ID.make("grp_session_one"), [session])
99+
const event = yield* bus.subscribe(Group.Event.ItemAdded).pipe(Stream.runHead, Effect.forkScoped)
100+
yield* Effect.yieldNow
101+
102+
yield* groups.addItem(group.id, terminal)
103+
104+
expect((yield* Fiber.join(event)).valueOrUndefined?.data).toEqual({ groupID: group.id, item: terminal })
105+
}),
106+
)
107+
})

‎packages/schema/src/event-manifest.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import { Event } from "./event.js"
1010
import { FileSystem } from "./filesystem.js"
1111
import { FileSystemV1 } from "./filesystem-v1.js"
1212
import { Form } from "./form.js"
13+
import { Group } from "./group.js"
1314
import { InstallationEvent } from "./installation-event.js"
1415
import { Integration } from "./integration.js"
1516
import { LegacyEventV1 } from "./legacy-event.js"
@@ -56,6 +57,7 @@ const featureDefinitions = Event.inventory(
5657
...Pty.Event.Definitions,
5758
...Shell.Event.Definitions,
5859
...Form.Event.Definitions,
60+
...Group.Event.Definitions,
5961
...WebSearch.Event.Definitions,
6062
)
6163

‎packages/schema/src/group.ts‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
export * as Group from "./group.js"
2+
3+
import { Schema } from "effect"
4+
import { ephemeral, inventory } from "./event.js"
5+
import { ascending } from "./identifier.js"
6+
import { Pty } from "./pty.js"
7+
import { statics } from "./schema.js"
8+
import { Session } from "./session.js"
9+
10+
const IDSchema = Schema.String.check(Schema.isStartsWith("grp_")).pipe(Schema.brand("GroupID"))
11+
12+
export const ID = IDSchema.pipe(
13+
statics((schema: typeof IDSchema) => ({ create: () => schema.make("grp_" + ascending()) })),
14+
)
15+
export type ID = typeof ID.Type
16+
17+
export const SessionItem = Schema.Struct({
18+
type: Schema.tag("session"),
19+
id: Session.ID,
20+
})
21+
export interface SessionItem extends Schema.Schema.Type<typeof SessionItem> {}
22+
23+
export const TerminalItem = Schema.Struct({
24+
type: Schema.tag("terminal"),
25+
id: Pty.ID,
26+
})
27+
export interface TerminalItem extends Schema.Schema.Type<typeof TerminalItem> {}
28+
29+
export const Item = Schema.Union([SessionItem, TerminalItem]).pipe(
30+
Schema.toTaggedUnion("type"),
31+
Schema.annotate({ identifier: "Group.Item" }),
32+
)
33+
export type Item = typeof Item.Type
34+
35+
export const Info = Schema.Struct({
36+
id: ID,
37+
items: Schema.Array(Item),
38+
}).annotate({ identifier: "Group.Info" })
39+
export interface Info extends Schema.Schema.Type<typeof Info> {}
40+
41+
const ItemAdded = ephemeral({ type: "group.item.added", schema: { groupID: ID, item: Item } })
42+
const ItemRemoved = ephemeral({ type: "group.item.removed", schema: { groupID: ID, item: Item } })
43+
export const Event = { ItemAdded, ItemRemoved, Definitions: inventory(ItemAdded, ItemRemoved) }

‎packages/schema/src/index.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ export { Credential } from "./credential.js"
66
export { Event } from "./event.js"
77
export { FileSystem } from "./filesystem.js"
88
export { Form } from "./form.js"
9+
export { Group } from "./group.js"
910
export { Integration } from "./integration.js"
1011
export { LLM } from "./llm.js"
1112
export { Location } from "./location.js"

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import {
44
Config,
55
FileSystem,
66
Form,
7+
Group,
78
Integration,
89
Permission,
910
Project,
@@ -64,6 +65,7 @@ describe("public event manifest", () => {
6465
expect(Integration.Event.Definitions).toEqual([Integration.Event.Updated, Integration.Event.ConnectionUpdated])
6566
expect(Permission.Event.Definitions).toEqual([Permission.Event.Asked, Permission.Event.Replied])
6667
expect(Form.Event.Definitions).toEqual([Form.Event.Created, Form.Event.Replied, Form.Event.Cancelled])
68+
expect(Group.Event.Definitions).toEqual([Group.Event.ItemAdded, Group.Event.ItemRemoved])
6769
expect(Reference.Event.Definitions).toEqual([Reference.Event.Updated])
6870
expect(Plugin.Event.Definitions).toEqual([Plugin.Event.Added, Plugin.Event.Updated])
6971
expect(McpEvent.Definitions).toEqual([McpEvent.ToolsChanged, McpEvent.ResourcesChanged, McpEvent.StatusChanged])

‎packages/schema/test/group.test.ts‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
import { describe, expect, test } from "bun:test"
2+
import { Schema } from "effect"
3+
import { Group } from "../src/group.js"
4+
import { Pty } from "../src/pty.js"
5+
import { Session } from "../src/session.js"
6+
7+
describe("Group", () => {
8+
test("creates branded group IDs", () => {
9+
expect(Group.ID.create()).toStartWith("grp_")
10+
expect(() => Schema.decodeUnknownSync(Group.ID)("ses_invalid")).toThrow()
11+
})
12+
13+
test("preserves one ordered session and terminal item list", () => {
14+
const group = Schema.decodeUnknownSync(Group.Info)({
15+
id: Group.ID.create(),
16+
items: [
17+
{ type: "session", id: Session.ID.make("ses_one") },
18+
{ type: "terminal", id: Pty.ID.make("pty_one") },
19+
{ type: "session", id: Session.ID.make("ses_two") },
20+
],
21+
})
22+
23+
expect(group.items.map((item) => item.type)).toEqual(["session", "terminal", "session"])
24+
expect(() =>
25+
Schema.decodeUnknownSync(Group.Info)({
26+
id: group.id,
27+
items: [{ type: "other", id: "other_one" }],
28+
}),
29+
).toThrow()
30+
})
31+
})

0 commit comments

Comments
 (0)