vrr/opencode
2
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-07-10 01:58:31 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

chore: generate

Browse source
This commit is contained in:
opencode-agent[bot] 2026-06-24 23:43:02 +00:00
parent 9bb5370205
commit 3730125f8f
14 changed files with 1587 additions and 492 deletions
Download patch file Download diff file Expand all files Collapse all files

20
packages/core/src/control-plane/move-session.ts
View file

@ -96,8 +96,8 @@ export const layer = Layer.effect(
if (patch) {
const repository = yield* git.repo.discover(directory)
if (!repository) return yield* new ApplyChangesError({ message: "Destination is not a Git repository" })
yield* git
.change.apply({ repository, path: directory, changes: patch })
yield* git.change
.apply({ repository, path: directory, changes: patch })
.pipe(Effect.mapError((error) => new ApplyChangesError({ message: error.message })))
}
@ -123,14 +123,14 @@ export const layer = Layer.effect(
untracked: "remove",
})
.pipe(
Effect.mapError(
(error) =>
new ResetSourceChangesError({
directory: current.location.directory,
message: error.message,
cause: error.cause,
}),
),
Effect.mapError(
(error) =>
new ResetSourceChangesError({
directory: current.location.directory,
message: error.message,
cause: error.cause,
}),
),
)
}
})

359
packages/core/src/git.ts
View file

@ -244,7 +244,10 @@ export const layer = Layer.effect(
directory: AbsolutePath,
args: string[],
) {
const result = yield* execute(directory, proc)(args).pipe(
const result = yield* execute(
directory,
proc,
)(args).pipe(
Effect.mapError((cause) => new OperationError({ operation, directory, message: cause.message, cause })),
)
if (result.exitCode === 0) return
@ -306,10 +309,7 @@ export const layer = Layer.effect(
])
})
const reset = Effect.fn("Git.sync.resetHard")(function* (
repository: Repository,
revision: string,
) {
const reset = Effect.fn("Git.sync.resetHard")(function* (repository: Repository, revision: string) {
yield* operation("reset", repository.worktree, ["reset", "--hard", revision])
})
@ -404,20 +404,22 @@ export const layer = Layer.effect(
}),
),
)
yield* fs.writeFileString(
path.join(input.gitDirectory, "objects", "info", "alternates"),
path.join(input.seed.commonDirectory, "objects") + "\n",
).pipe(
Effect.mapError(
(cause) =>
new OperationError({
operation: "create",
directory: input.gitDirectory,
message: "Failed to configure shared Git objects",
cause,
}),
),
)
yield* fs
.writeFileString(
path.join(input.gitDirectory, "objects", "info", "alternates"),
path.join(input.seed.commonDirectory, "objects") + "\n",
)
.pipe(
Effect.mapError(
(cause) =>
new OperationError({
operation: "create",
directory: input.gitDirectory,
message: "Failed to configure shared Git objects",
cause,
}),
),
)
yield* fs
.copyFile(path.join(input.seed.gitDirectory, "index"), path.join(input.gitDirectory, "index"))
.pipe(Effect.catch(() => Effect.void))
@ -445,14 +447,9 @@ export const layer = Layer.effect(
if (!candidates.length) return { skipped: [] }
const ignored = input.ignores
? new Set(
(
yield* repositoryOperation(
"refresh",
input.ignores,
["check-ignore", "--no-index", "--stdin", "-z"],
{ stdin: candidates.join("\0") + "\0" },
).pipe(Effect.catch(() => Effect.succeed({ text: "", stderr: "" })))
).text
(yield* repositoryOperation("refresh", input.ignores, ["check-ignore", "--no-index", "--stdin", "-z"], {
stdin: candidates.join("\0") + "\0",
}).pipe(Effect.catch(() => Effect.succeed({ text: "", stderr: "" })))).text
.split("\0")
.filter(Boolean),
)
@ -460,19 +457,17 @@ export const layer = Layer.effect(
const allowed = candidates.filter((item) => !ignored.has(item))
const maximum = input.maximumUntrackedFileBytes
const skipped = maximum
? (
yield* Effect.forEach(
untracked.filter((item) => allowed.includes(item)),
(item) =>
fs.stat(path.join(input.repository.worktree, item)).pipe(
Effect.map((info) =>
info.type === "File" && Number(info.size) > maximum ? RelativePath.make(item) : undefined,
),
Effect.catch(() => Effect.succeed(undefined)),
? (yield* Effect.forEach(
untracked.filter((item) => allowed.includes(item)),
(item) =>
fs.stat(path.join(input.repository.worktree, item)).pipe(
Effect.map((info) =>
info.type === "File" && Number(info.size) > maximum ? RelativePath.make(item) : undefined,
),
{ concurrency: 8 },
)
).filter((item): item is RelativePath => item !== undefined)
Effect.catch(() => Effect.succeed(undefined)),
),
{ concurrency: 8 },
)).filter((item): item is RelativePath => item !== undefined)
: []
const stage = allowed.filter((item) => !skipped.includes(RelativePath.make(item)))
const remove = [...ignored, ...skipped]
@ -500,11 +495,10 @@ export const layer = Layer.effect(
if (!input.paths.length) return new Set<RelativePath>()
const result = yield* proc
.run(
ChildProcess.make(
"git",
repositoryArgs(input.repository, ["check-ignore", "--no-index", "--stdin", "-z"]),
{ cwd: input.repository.worktree, extendEnv: true },
),
ChildProcess.make("git", repositoryArgs(input.repository, ["check-ignore", "--no-index", "--stdin", "-z"]), {
cwd: input.repository.worktree,
extendEnv: true,
}),
{ stdin: input.paths.join("\0") + "\0" },
)
.pipe(
@ -537,23 +531,20 @@ export const layer = Layer.effect(
return TreeID.make((yield* repositoryOperation("write_tree", repository, ["write-tree"])).text.trim())
})
const captureTree = Effect.fn("Git.tree.capture")((input: {
repository: Repository
scopes: readonly RelativePath[]
ignores?: Repository
maximumUntrackedFileBytes?: number
}) =>
locked(
input.repository,
Effect.gen(function* () {
yield* Effect.forEach(
input.scopes,
(scope) => refresh({ ...input, scope }),
{ discard: true },
)
return yield* writeTree(input.repository)
}),
),
const captureTree = Effect.fn("Git.tree.capture")(
(input: {
repository: Repository
scopes: readonly RelativePath[]
ignores?: Repository
maximumUntrackedFileBytes?: number
}) =>
locked(
input.repository,
Effect.gen(function* () {
yield* Effect.forEach(input.scopes, (scope) => refresh({ ...input, scope }), { discard: true })
return yield* writeTree(input.repository)
}),
),
)
const treeFiles = Effect.fn("Git.tree.files")(function* (input: {
@ -561,8 +552,14 @@ export const layer = Layer.effect(
from: TreeID
to: TreeID
}) {
return (yield* repositoryOperation("list_files", input.repository, ["diff", "--name-only", "-z", input.from, input.to]))
.text.split("\0")
return (yield* repositoryOperation("list_files", input.repository, [
"diff",
"--name-only",
"-z",
input.from,
input.to,
])).text
.split("\0")
.filter(Boolean)
.map((file) => RelativePath.make(file))
})
@ -577,43 +574,37 @@ export const layer = Layer.effect(
const paths = input.paths ?? (yield* treeFiles(input))
return yield* Effect.forEach(paths, (file) =>
Effect.gen(function* () {
const statusText = (
yield* repositoryOperation("diff", input.repository, [
"diff",
"--name-status",
"--no-renames",
input.from,
input.to,
"--",
file,
])
).text.trim()
const statusText = (yield* repositoryOperation("diff", input.repository, [
"diff",
"--name-status",
"--no-renames",
input.from,
input.to,
"--",
file,
])).text.trim()
const status = statusText.startsWith("A") ? "added" : statusText.startsWith("D") ? "deleted" : "modified"
const stats = (
yield* repositoryOperation("diff", input.repository, [
"diff",
"--numstat",
"--no-renames",
input.from,
input.to,
"--",
file,
])
).text.split("\t")
const stats = (yield* repositoryOperation("diff", input.repository, [
"diff",
"--numstat",
"--no-renames",
input.from,
input.to,
"--",
file,
])).text.split("\t")
const binary = stats[0] === "-" || stats[1] === "-"
const patch = binary
? ""
: (
yield* repositoryOperation("diff", input.repository, [
"diff",
`--unified=${input.context ?? 3}`,
"--no-renames",
input.from,
input.to,
"--",
file,
])
).text
: (yield* repositoryOperation("diff", input.repository, [
"diff",
`--unified=${input.context ?? 3}`,
"--no-renames",
input.from,
input.to,
"--",
file,
])).text
return {
path: file,
status,
@ -626,87 +617,89 @@ export const layer = Layer.effect(
})
const entry = Effect.fnUntraced(function* (repository: Repository, tree: TreeID, file: RelativePath) {
const text = (
yield* repositoryOperation("restore", repository, ["ls-tree", "-z", tree, "--", file])
).text.replace(/\0$/, "")
const text = (yield* repositoryOperation("restore", repository, [
"ls-tree",
"-z",
tree,
"--",
file,
])).text.replace(/\0$/, "")
if (!text) return
const match = text.match(/^(\d+)\s+\w+\s+([0-9a-f]+)\t/)
if (!match) return yield* new OperationError({
operation: "restore",
directory: repository.worktree,
message: `Invalid tree entry for ${file}`,
})
if (!match)
return yield* new OperationError({
operation: "restore",
directory: repository.worktree,
message: `Invalid tree entry for ${file}`,
})
return { mode: match[1], object: match[2] }
})
const preview = Effect.fn("Git.tree.preview")((input: {
repository: Repository
current: TreeID
files: ReadonlyMap<RelativePath, TreeID>
context?: number
}) =>
locked(
input.repository,
Effect.gen(function* () {
const index = path.join(input.repository.gitDirectory, `preview-${randomUUID()}.index`)
const env = { GIT_INDEX_FILE: index }
return yield* Effect.gen(function* () {
yield* repositoryOperation("diff", input.repository, ["read-tree", input.current], { env })
yield* Effect.forEach(
input.files,
([file, tree]) =>
Effect.gen(function* () {
const source = yield* entry(input.repository, tree, file)
if (!source) {
const preview = Effect.fn("Git.tree.preview")(
(input: {
repository: Repository
current: TreeID
files: ReadonlyMap<RelativePath, TreeID>
context?: number
}) =>
locked(
input.repository,
Effect.gen(function* () {
const index = path.join(input.repository.gitDirectory, `preview-${randomUUID()}.index`)
const env = { GIT_INDEX_FILE: index }
return yield* Effect.gen(function* () {
yield* repositoryOperation("diff", input.repository, ["read-tree", input.current], { env })
yield* Effect.forEach(
input.files,
([file, tree]) =>
Effect.gen(function* () {
const source = yield* entry(input.repository, tree, file)
if (!source) {
yield* repositoryOperation(
"diff",
input.repository,
["update-index", "--force-remove", "--", file],
{ env },
)
return
}
yield* repositoryOperation(
"diff",
input.repository,
["update-index", "--force-remove", "--", file],
["update-index", "--add", "--cacheinfo", source.mode, source.object, file],
{ env },
)
return
}
yield* repositoryOperation(
"diff",
input.repository,
["update-index", "--add", "--cacheinfo", source.mode, source.object, file],
{ env },
)
}),
{ discard: true },
)
const target = TreeID.make(
(yield* repositoryOperation("diff", input.repository, ["write-tree"], { env })).text.trim(),
)
return yield* treeDiff({
repository: input.repository,
from: input.current,
to: target,
context: input.context,
paths: Array.from(input.files.keys()),
})
}).pipe(Effect.ensuring(fs.remove(index).pipe(Effect.catch(() => Effect.void))))
}),
),
}),
{ discard: true },
)
const target = TreeID.make(
(yield* repositoryOperation("diff", input.repository, ["write-tree"], { env })).text.trim(),
)
return yield* treeDiff({
repository: input.repository,
from: input.current,
to: target,
context: input.context,
paths: Array.from(input.files.keys()),
})
}).pipe(Effect.ensuring(fs.remove(index).pipe(Effect.catch(() => Effect.void))))
}),
),
)
const restore = Effect.fn("Git.tree.restore")((input: {
repository: Repository
files: ReadonlyMap<RelativePath, TreeID>
}) =>
locked(
input.repository,
Effect.forEach(
input.files,
([file, tree]) =>
Effect.gen(function* () {
if (yield* entry(input.repository, tree, file)) {
yield* repositoryOperation("restore", input.repository, ["checkout", tree, "--", file])
return
}
yield* fs
.remove(path.join(input.repository.worktree, file), { recursive: true, force: true })
.pipe(
const restore = Effect.fn("Git.tree.restore")(
(input: { repository: Repository; files: ReadonlyMap<RelativePath, TreeID> }) =>
locked(
input.repository,
Effect.forEach(
input.files,
([file, tree]) =>
Effect.gen(function* () {
if (yield* entry(input.repository, tree, file)) {
yield* repositoryOperation("restore", input.repository, ["checkout", tree, "--", file])
return
}
yield* fs.remove(path.join(input.repository.worktree, file), { recursive: true, force: true }).pipe(
Effect.mapError(
(cause) =>
new OperationError({
@ -717,10 +710,10 @@ export const layer = Layer.effect(
}),
),
)
}),
{ discard: true },
}),
{ discard: true },
),
),
),
)
const checkoutTree = Effect.fn("Git.tree.checkout")((input: { repository: Repository; tree: TreeID }) =>
@ -733,10 +726,7 @@ export const layer = Layer.effect(
),
)
const capture = Effect.fn("Git.change.capture")(function* (input: {
repository: Repository
path: AbsolutePath
}) {
const capture = Effect.fn("Git.change.capture")(function* (input: { repository: Repository; path: AbsolutePath }) {
const scope = path.relative(input.repository.worktree, input.path).replaceAll("\\", "/") || "."
const tracked = yield* execute(
input.repository.worktree,
@ -811,8 +801,7 @@ export const layer = Layer.effect(
)
.pipe(
Effect.mapError(
(cause) =>
new PatchError({ operation: "apply", directory: input.path, message: cause.message, cause }),
(cause) => new PatchError({ operation: "apply", directory: input.path, message: cause.message, cause }),
),
)
if (result.exitCode === 0) return
@ -831,9 +820,10 @@ export const layer = Layer.effect(
untracked: "preserve" | "remove"
}) {
const scope = path.relative(input.repository.worktree, input.path).replaceAll("\\", "/") || "."
const restore = yield* execute(input.repository.worktree, proc)(
input.index === "reset" ? ["checkout", "HEAD", "--", scope] : ["checkout", "--", scope],
).pipe(
const restore = yield* execute(
input.repository.worktree,
proc,
)(input.index === "reset" ? ["checkout", "HEAD", "--", scope] : ["checkout", "--", scope]).pipe(
Effect.mapError(
(cause) => new PatchError({ operation: "reset", directory: input.path, message: cause.message, cause }),
),
@ -846,7 +836,10 @@ export const layer = Layer.effect(
})
}
if (input.untracked === "preserve") return
const clean = yield* execute(input.repository.worktree, proc)(["clean", "-fd", "--", scope]).pipe(
const clean = yield* execute(
input.repository.worktree,
proc,
)(["clean", "-fd", "--", scope]).pipe(
Effect.mapError(
(cause) => new PatchError({ operation: "reset", directory: input.path, message: cause.message, cause }),
),
@ -937,7 +930,15 @@ export const layer = Layer.effect(
change: { capture, apply, discard },
worktree: { create: worktreeCreate, remove: worktreeRemove, list: worktreeList },
index: { refresh, ignored },
tree: { capture: captureTree, write: writeTree, files: treeFiles, diff: treeDiff, preview, restore, checkout: checkoutTree },
tree: {
capture: captureTree,
write: writeTree,
files: treeFiles,
diff: treeDiff,
preview,
restore,
checkout: checkoutTree,
},
})
}),
)

3
packages/core/src/repository-cache.ts
View file

@ -169,7 +169,8 @@ export const layer: Layer.Layer<Service, never, FSUtil.Service | Git.Service | E
}
if (status === "refreshed") {
if (!existing) return yield* new FetchFailedError({ repository, message: "Repository is unavailable" })
if (!existing)
return yield* new FetchFailedError({ repository, message: "Repository is unavailable" })
yield* git.sync
.fetchRemotes(existing)
.pipe(Effect.mapError((error) => new FetchFailedError({ repository, message: error.message })))

514
packages/core/src/session.ts
View file

@ -172,261 +172,265 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/v2
export const layer = Layer.unwrap(
Effect.promise(() => import("./location-layer")).pipe(
Effect.map(({ LocationServiceMap }) => Layer.effect(
Service,
Effect.gen(function* () {
const database = yield* Database.Service
const db = database.db
const events = yield* EventV2.Service
const projects = yield* ProjectV2.Service
const execution = yield* SessionExecution.Service
const store = yield* SessionStore.Service
const locations = yield* LocationServiceMap
const decodeMessage = Schema.decodeUnknownEffect(SessionMessage.Message)
const isDurableSessionEvent = Schema.is(SessionEvent.Durable)
const decode = (row: typeof SessionMessageTable.$inferSelect) =>
decodeMessage({ ...row.data, id: row.id, type: row.type }).pipe(
Effect.mapError(
() =>
new MessageDecodeError({
sessionID: SessionSchema.ID.make(row.session_id),
messageID: SessionMessage.ID.make(row.id),
}),
),
)
const result = Service.of({
create: Effect.fn("V2Session.create")(function* (input) {
const sessionID = input.id ?? SessionSchema.ID.create()
const recorded = yield* store.get(sessionID)
if (recorded) return recorded
const project = yield* projects.resolve(input.location.directory)
yield* db
.insert(ProjectTable)
.values({ id: project.id, worktree: project.directory, vcs: project.vcs?.type, sandboxes: [] })
.onConflictDoNothing()
.run()
.pipe(Effect.orDie)
const now = Date.now()
const info = SessionV1.SessionInfo.make({
id: sessionID,
slug: Slug.create(),
version: InstallationVersion,
projectID: project.id,
directory: input.location.directory,
path: path.relative(project.directory, input.location.directory).replaceAll("\\", "/"),
workspaceID: input.location.workspaceID ? WorkspaceV2.ID.make(input.location.workspaceID) : undefined,
title: `New session - ${new Date(now).toISOString()}`,
agent: input.agent,
model: input.model
? {
id: ModelV2.ID.make(input.model.id),
providerID: input.model.providerID,
variant: input.model.variant,
}
: undefined,
cost: 0,
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
time: { created: now, updated: now },
})
const projected = yield* events
.publish(SessionV1.Event.Created, { sessionID, info }, { location: input.location })
.pipe(
Effect.as({ type: "created" } as const),
Effect.catchDefect((defect) => {
if (!(defect instanceof SessionProjector.SessionAlreadyProjected)) {
return Effect.die(defect)
}
// Concurrent creation lost the projection race. The existing Session identity wins.
return store
.get(sessionID)
.pipe(
Effect.flatMap((session) =>
session ? Effect.succeed({ type: "existing", session } as const) : Effect.die(defect),
),
)
}),
)
if (projected.type === "existing") return projected.session
// TODO: Restore recorded sessions onto replacement synchronized workspaces in a future API slice.
return yield* result.get(sessionID).pipe(Effect.orDie)
}),
get: Effect.fn("V2Session.get")(function* (sessionID) {
const session = yield* store.get(sessionID)
if (!session) return yield* new NotFoundError({ sessionID })
return session
}),
list: Effect.fn("V2Session.list")(function* (input = {}) {
const direction = input.anchor?.direction ?? "next"
const requestedOrder = input.order ?? "desc"
const order = direction === "previous" ? (requestedOrder === "asc" ? "desc" : "asc") : requestedOrder
const sortColumn = SessionTable.time_created
const conditions: SQL[] = []
if ("directory" in input) conditions.push(eq(SessionTable.directory, input.directory))
if (input.workspaceID) conditions.push(eq(SessionTable.workspace_id, input.workspaceID))
if ("project" in input) conditions.push(eq(SessionTable.project_id, input.project))
if (input.search) conditions.push(like(SessionTable.title, `%${input.search}%`))
if (input.anchor) {
conditions.push(
order === "asc"
? or(
gt(sortColumn, input.anchor.time),
and(eq(sortColumn, input.anchor.time), gt(SessionTable.id, input.anchor.id)),
)!
: or(
lt(sortColumn, input.anchor.time),
and(eq(sortColumn, input.anchor.time), lt(SessionTable.id, input.anchor.id)),
)!,
)
}
const query = db
.select()
.from(SessionTable)
.where(conditions.length > 0 ? and(...conditions) : undefined)
.orderBy(
order === "asc" ? asc(sortColumn) : desc(sortColumn),
order === "asc" ? asc(SessionTable.id) : desc(SessionTable.id),
)
const rows = yield* (input.limit === undefined ? query.all() : query.limit(input.limit).all()).pipe(
Effect.orDie,
)
return (direction === "previous" ? rows.toReversed() : rows).map((row) => fromRow(row))
}),
messages: Effect.fn("V2Session.messages")(function* (input) {
yield* result.get(input.sessionID)
const direction = input.cursor?.direction ?? "next"
const requestedOrder = input.order ?? "desc"
const order = direction === "previous" ? (requestedOrder === "asc" ? "desc" : "asc") : requestedOrder
const anchor = input.cursor
? yield* db
.select({ seq: SessionMessageTable.seq })
.from(SessionMessageTable)
.where(
and(eq(SessionMessageTable.session_id, input.sessionID), eq(SessionMessageTable.id, input.cursor.id)),
)
.get()
.pipe(Effect.orDie)
: undefined
if (input.cursor && !anchor) return []
const boundary = anchor
? order === "asc"
? gt(SessionMessageTable.seq, anchor.seq)
: lt(SessionMessageTable.seq, anchor.seq)
: undefined
const where = boundary
? and(eq(SessionMessageTable.session_id, input.sessionID), boundary)
: eq(SessionMessageTable.session_id, input.sessionID)
const query = db
.select()
.from(SessionMessageTable)
.where(where)
.orderBy(order === "asc" ? asc(SessionMessageTable.seq) : desc(SessionMessageTable.seq))
const rows = yield* (input.limit === undefined ? query.all() : query.limit(input.limit).all()).pipe(
Effect.orDie,
)
return yield* Effect.forEach(direction === "previous" ? rows.toReversed() : rows, decode)
}),
message: Effect.fn("V2Session.message")(function* (input) {
const stored = yield* store.message(input.messageID)
return stored?.sessionID === input.sessionID ? stored.message : undefined
}),
context: Effect.fn("V2Session.context")(function* (sessionID) {
yield* result.get(sessionID)
return yield* store.context(sessionID)
}),
events: (input) =>
Stream.unwrap(
result
.get(input.sessionID)
.pipe(Effect.as(events.durable({ aggregateID: input.sessionID, after: input.after }))),
).pipe(Stream.filter((event): event is SessionEvent.DurableEvent => isDurableSessionEvent(event))),
prompt: Effect.fn("V2Session.prompt")((input) =>
Effect.uninterruptible(
Effect.gen(function* () {
yield* result.get(input.sessionID)
const messageID = input.id ?? SessionMessage.ID.create()
const delivery = input.delivery ?? "steer"
const expected = { sessionID: input.sessionID, messageID, prompt: input.prompt, delivery }
const admitted = yield* SessionInput.admit(db, events, {
id: messageID,
sessionID: input.sessionID,
prompt: input.prompt,
delivery,
}).pipe(
Effect.catchDefect((defect) =>
defect instanceof SessionInput.LifecycleConflict
? new PromptConflictError({ sessionID: input.sessionID, messageID })
: Effect.die(defect),
Effect.map(({ LocationServiceMap }) =>
Layer.effect(
Service,
Effect.gen(function* () {
const database = yield* Database.Service
const db = database.db
const events = yield* EventV2.Service
const projects = yield* ProjectV2.Service
const execution = yield* SessionExecution.Service
const store = yield* SessionStore.Service
const locations = yield* LocationServiceMap
const decodeMessage = Schema.decodeUnknownEffect(SessionMessage.Message)
const isDurableSessionEvent = Schema.is(SessionEvent.Durable)
const decode = (row: typeof SessionMessageTable.$inferSelect) =>
decodeMessage({ ...row.data, id: row.id, type: row.type }).pipe(
Effect.mapError(
() =>
new MessageDecodeError({
sessionID: SessionSchema.ID.make(row.session_id),
messageID: SessionMessage.ID.make(row.id),
}),
),
)
if (!SessionInput.equivalent(admitted, expected))
return yield* new PromptConflictError({ sessionID: input.sessionID, messageID })
if (input.resume !== false) yield* execution.wake(admitted.sessionID)
return admitted
}),
),
),
shell: Effect.fn("V2Session.shell")(function* () {
return yield* new OperationUnavailableError({ operation: "shell" })
}),
skill: Effect.fn("V2Session.skill")(function* () {
return yield* new OperationUnavailableError({ operation: "skill" })
}),
switchAgent: Effect.fn("V2Session.switchAgent")(function* (input) {
yield* result.get(input.sessionID)
yield* events.publish(SessionEvent.AgentSwitched, {
sessionID: input.sessionID,
messageID: SessionMessage.ID.create(),
timestamp: yield* DateTime.now,
agent: input.agent,
})
}),
switchModel: Effect.fn("V2Session.switchModel")(function* (input) {
yield* result.get(input.sessionID)
yield* events.publish(SessionEvent.ModelSwitched, {
sessionID: input.sessionID,
messageID: SessionMessage.ID.create(),
timestamp: yield* DateTime.now,
model: input.model,
})
}),
compact: Effect.fn("V2Session.compact")(function* (input) {
yield* result.get(input.sessionID)
return yield* new OperationUnavailableError({ operation: "compact" })
}),
wait: Effect.fn("V2Session.wait")(function* (sessionID) {
yield* result.get(sessionID)
return yield* new OperationUnavailableError({ operation: "wait" })
}),
resume: Effect.fn("V2Session.resume")(function* (sessionID) {
yield* result.get(sessionID)
yield* execution.resume(sessionID)
}),
interrupt: Effect.fn("V2Session.interrupt")((sessionID) =>
Effect.uninterruptible(execution.interrupt(sessionID)),
),
revert: {
stage: Effect.fn("V2Session.revert.stage")(function* (input) {
const session = yield* result.get(input.sessionID)
return yield* SessionRevert.stage({ session, messageID: input.messageID, files: input.files }).pipe(
Effect.provideService(Database.Service, database),
Effect.provideService(EventV2.Service, events),
Effect.provide(locations.get(session.location)),
)
}),
clear: Effect.fn("V2Session.revert.clear")(function* (sessionID) {
const session = yield* result.get(sessionID)
yield* SessionRevert.clear(session).pipe(
Effect.provideService(EventV2.Service, events),
Effect.provide(locations.get(session.location)),
)
}),
commit: Effect.fn("V2Session.revert.commit")(function* (sessionID) {
const session = yield* result.get(sessionID)
yield* SessionRevert.commit(session).pipe(Effect.provideService(EventV2.Service, events))
}),
},
})
const result = Service.of({
create: Effect.fn("V2Session.create")(function* (input) {
const sessionID = input.id ?? SessionSchema.ID.create()
const recorded = yield* store.get(sessionID)
if (recorded) return recorded
const project = yield* projects.resolve(input.location.directory)
yield* db
.insert(ProjectTable)
.values({ id: project.id, worktree: project.directory, vcs: project.vcs?.type, sandboxes: [] })
.onConflictDoNothing()
.run()
.pipe(Effect.orDie)
const now = Date.now()
const info = SessionV1.SessionInfo.make({
id: sessionID,
slug: Slug.create(),
version: InstallationVersion,
projectID: project.id,
directory: input.location.directory,
path: path.relative(project.directory, input.location.directory).replaceAll("\\", "/"),
workspaceID: input.location.workspaceID ? WorkspaceV2.ID.make(input.location.workspaceID) : undefined,
title: `New session - ${new Date(now).toISOString()}`,
agent: input.agent,
model: input.model
? {
id: ModelV2.ID.make(input.model.id),
providerID: input.model.providerID,
variant: input.model.variant,
}
: undefined,
cost: 0,
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
time: { created: now, updated: now },
})
const projected = yield* events
.publish(SessionV1.Event.Created, { sessionID, info }, { location: input.location })
.pipe(
Effect.as({ type: "created" } as const),
Effect.catchDefect((defect) => {
if (!(defect instanceof SessionProjector.SessionAlreadyProjected)) {
return Effect.die(defect)
}
// Concurrent creation lost the projection race. The existing Session identity wins.
return store
.get(sessionID)
.pipe(
Effect.flatMap((session) =>
session ? Effect.succeed({ type: "existing", session } as const) : Effect.die(defect),
),
)
}),
)
if (projected.type === "existing") return projected.session
// TODO: Restore recorded sessions onto replacement synchronized workspaces in a future API slice.
return yield* result.get(sessionID).pipe(Effect.orDie)
}),
get: Effect.fn("V2Session.get")(function* (sessionID) {
const session = yield* store.get(sessionID)
if (!session) return yield* new NotFoundError({ sessionID })
return session
}),
list: Effect.fn("V2Session.list")(function* (input = {}) {
const direction = input.anchor?.direction ?? "next"
const requestedOrder = input.order ?? "desc"
const order = direction === "previous" ? (requestedOrder === "asc" ? "desc" : "asc") : requestedOrder
const sortColumn = SessionTable.time_created
const conditions: SQL[] = []
if ("directory" in input) conditions.push(eq(SessionTable.directory, input.directory))
if (input.workspaceID) conditions.push(eq(SessionTable.workspace_id, input.workspaceID))
if ("project" in input) conditions.push(eq(SessionTable.project_id, input.project))
if (input.search) conditions.push(like(SessionTable.title, `%${input.search}%`))
if (input.anchor) {
conditions.push(
order === "asc"
? or(
gt(sortColumn, input.anchor.time),
and(eq(sortColumn, input.anchor.time), gt(SessionTable.id, input.anchor.id)),
)!
: or(
lt(sortColumn, input.anchor.time),
and(eq(sortColumn, input.anchor.time), lt(SessionTable.id, input.anchor.id)),
)!,
)
}
const query = db
.select()
.from(SessionTable)
.where(conditions.length > 0 ? and(...conditions) : undefined)
.orderBy(
order === "asc" ? asc(sortColumn) : desc(sortColumn),
order === "asc" ? asc(SessionTable.id) : desc(SessionTable.id),
)
const rows = yield* (input.limit === undefined ? query.all() : query.limit(input.limit).all()).pipe(
Effect.orDie,
)
return (direction === "previous" ? rows.toReversed() : rows).map((row) => fromRow(row))
}),
messages: Effect.fn("V2Session.messages")(function* (input) {
yield* result.get(input.sessionID)
const direction = input.cursor?.direction ?? "next"
const requestedOrder = input.order ?? "desc"
const order = direction === "previous" ? (requestedOrder === "asc" ? "desc" : "asc") : requestedOrder
const anchor = input.cursor
? yield* db
.select({ seq: SessionMessageTable.seq })
.from(SessionMessageTable)
.where(
and(
eq(SessionMessageTable.session_id, input.sessionID),
eq(SessionMessageTable.id, input.cursor.id),
),
)
.get()
.pipe(Effect.orDie)
: undefined
if (input.cursor && !anchor) return []
const boundary = anchor
? order === "asc"
? gt(SessionMessageTable.seq, anchor.seq)
: lt(SessionMessageTable.seq, anchor.seq)
: undefined
const where = boundary
? and(eq(SessionMessageTable.session_id, input.sessionID), boundary)
: eq(SessionMessageTable.session_id, input.sessionID)
const query = db
.select()
.from(SessionMessageTable)
.where(where)
.orderBy(order === "asc" ? asc(SessionMessageTable.seq) : desc(SessionMessageTable.seq))
const rows = yield* (input.limit === undefined ? query.all() : query.limit(input.limit).all()).pipe(
Effect.orDie,
)
return yield* Effect.forEach(direction === "previous" ? rows.toReversed() : rows, decode)
}),
message: Effect.fn("V2Session.message")(function* (input) {
const stored = yield* store.message(input.messageID)
return stored?.sessionID === input.sessionID ? stored.message : undefined
}),
context: Effect.fn("V2Session.context")(function* (sessionID) {
yield* result.get(sessionID)
return yield* store.context(sessionID)
}),
events: (input) =>
Stream.unwrap(
result
.get(input.sessionID)
.pipe(Effect.as(events.durable({ aggregateID: input.sessionID, after: input.after }))),
).pipe(Stream.filter((event): event is SessionEvent.DurableEvent => isDurableSessionEvent(event))),
prompt: Effect.fn("V2Session.prompt")((input) =>
Effect.uninterruptible(
Effect.gen(function* () {
yield* result.get(input.sessionID)
const messageID = input.id ?? SessionMessage.ID.create()
const delivery = input.delivery ?? "steer"
const expected = { sessionID: input.sessionID, messageID, prompt: input.prompt, delivery }
const admitted = yield* SessionInput.admit(db, events, {
id: messageID,
sessionID: input.sessionID,
prompt: input.prompt,
delivery,
}).pipe(
Effect.catchDefect((defect) =>
defect instanceof SessionInput.LifecycleConflict
? new PromptConflictError({ sessionID: input.sessionID, messageID })
: Effect.die(defect),
),
)
if (!SessionInput.equivalent(admitted, expected))
return yield* new PromptConflictError({ sessionID: input.sessionID, messageID })
if (input.resume !== false) yield* execution.wake(admitted.sessionID)
return admitted
}),
),
),
shell: Effect.fn("V2Session.shell")(function* () {
return yield* new OperationUnavailableError({ operation: "shell" })
}),
skill: Effect.fn("V2Session.skill")(function* () {
return yield* new OperationUnavailableError({ operation: "skill" })
}),
switchAgent: Effect.fn("V2Session.switchAgent")(function* (input) {
yield* result.get(input.sessionID)
yield* events.publish(SessionEvent.AgentSwitched, {
sessionID: input.sessionID,
messageID: SessionMessage.ID.create(),
timestamp: yield* DateTime.now,
agent: input.agent,
})
}),
switchModel: Effect.fn("V2Session.switchModel")(function* (input) {
yield* result.get(input.sessionID)
yield* events.publish(SessionEvent.ModelSwitched, {
sessionID: input.sessionID,
messageID: SessionMessage.ID.create(),
timestamp: yield* DateTime.now,
model: input.model,
})
}),
compact: Effect.fn("V2Session.compact")(function* (input) {
yield* result.get(input.sessionID)
return yield* new OperationUnavailableError({ operation: "compact" })
}),
wait: Effect.fn("V2Session.wait")(function* (sessionID) {
yield* result.get(sessionID)
return yield* new OperationUnavailableError({ operation: "wait" })
}),
resume: Effect.fn("V2Session.resume")(function* (sessionID) {
yield* result.get(sessionID)
yield* execution.resume(sessionID)
}),
interrupt: Effect.fn("V2Session.interrupt")((sessionID) =>
Effect.uninterruptible(execution.interrupt(sessionID)),
),
revert: {
stage: Effect.fn("V2Session.revert.stage")(function* (input) {
const session = yield* result.get(input.sessionID)
return yield* SessionRevert.stage({ session, messageID: input.messageID, files: input.files }).pipe(
Effect.provideService(Database.Service, database),
Effect.provideService(EventV2.Service, events),
Effect.provide(locations.get(session.location)),
)
}),
clear: Effect.fn("V2Session.revert.clear")(function* (sessionID) {
const session = yield* result.get(sessionID)
yield* SessionRevert.clear(session).pipe(
Effect.provideService(EventV2.Service, events),
Effect.provide(locations.get(session.location)),
)
}),
commit: Effect.fn("V2Session.revert.commit")(function* (sessionID) {
const session = yield* result.get(sessionID)
yield* SessionRevert.commit(session).pipe(Effect.provideService(EventV2.Service, events))
}),
},
})
return result
}),
@ -436,7 +440,9 @@ export const layer = Layer.unwrap(
)
export const defaultLayer = layer.pipe(
Layer.provide(Layer.unwrap(Effect.promise(() => import("./location-layer")).pipe(Effect.map((m) => m.LocationServiceMap.layer)))),
Layer.provide(
Layer.unwrap(Effect.promise(() => import("./location-layer")).pipe(Effect.map((m) => m.LocationServiceMap.layer))),
),
Layer.provide(SessionExecution.noopLayer),
Layer.provide(SessionStore.defaultLayer),
Layer.provide(SessionProjector.defaultLayer),

4
packages/core/src/session/info.ts
View file

@ -40,9 +40,7 @@ export function fromRow(row: typeof SessionTable.$inferSelect): SessionSchema.In
workspaceID: row.workspace_id ? WorkspaceV2.ID.make(row.workspace_id) : undefined,
}),
subpath: row.path ? RelativePath.make(row.path) : undefined,
revert: row.revert
? { ...row.revert, messageID: SessionMessageID.ID.make(row.revert.messageID) }
: undefined,
revert: row.revert ? { ...row.revert, messageID: SessionMessageID.ID.make(row.revert.messageID) } : undefined,
time: {
created: DateTime.makeUnsafe(row.time_created),
updated: DateTime.makeUnsafe(row.time_updated),

33
packages/core/src/session/projector.ts
View file

@ -418,13 +418,38 @@ export const layer = Layer.effectDiscard(
const boundary = yield* db
.select({ seq: SessionMessageTable.seq })
.from(SessionMessageTable)
.where(and(eq(SessionMessageTable.session_id, event.data.sessionID), eq(SessionMessageTable.id, event.data.messageID)))
.where(
and(
eq(SessionMessageTable.session_id, event.data.sessionID),
eq(SessionMessageTable.id, event.data.messageID),
),
)
.get()
.pipe(Effect.orDie)
if (!boundary) return yield* Effect.die(`Revert boundary message not found: ${event.data.messageID}`)
yield* db.delete(SessionMessageTable).where(and(eq(SessionMessageTable.session_id, event.data.sessionID), gt(SessionMessageTable.seq, boundary.seq))).run().pipe(Effect.orDie)
yield* db.delete(SessionInputTable).where(and(eq(SessionInputTable.session_id, event.data.sessionID), or(gt(SessionInputTable.admitted_seq, boundary.seq), gt(SessionInputTable.promoted_seq, boundary.seq)))).run().pipe(Effect.orDie)
yield* db.update(SessionTable).set({ revert: null, time_updated: DateTime.toEpochMillis(event.data.timestamp) }).where(eq(SessionTable.id, event.data.sessionID)).run().pipe(Effect.orDie)
yield* db
.delete(SessionMessageTable)
.where(
and(eq(SessionMessageTable.session_id, event.data.sessionID), gt(SessionMessageTable.seq, boundary.seq)),
)
.run()
.pipe(Effect.orDie)
yield* db
.delete(SessionInputTable)
.where(
and(
eq(SessionInputTable.session_id, event.data.sessionID),
or(gt(SessionInputTable.admitted_seq, boundary.seq), gt(SessionInputTable.promoted_seq, boundary.seq)),
),
)
.run()
.pipe(Effect.orDie)
yield* db
.update(SessionTable)
.set({ revert: null, time_updated: DateTime.toEpochMillis(event.data.timestamp) })
.where(eq(SessionTable.id, event.data.sessionID))
.run()
.pipe(Effect.orDie)
yield* SessionContextEpoch.reset(db, event.data.sessionID)
}),
)

7
packages/core/src/session/revert.ts
View file

@ -66,7 +66,7 @@ export const stage = Effect.fn("SessionRevert.stage")(function* (input: {
const events = yield* EventV2.Service
const original = input.session.revert?.snapshot
? Snapshot.ID.make(input.session.revert.snapshot)
: (yield* snapshot.capture())
: yield* snapshot.capture()
const next = yield* plan({ sessionID: input.session.id, messageID: input.messageID })
const restore = new Map<RelativePath, Snapshot.ID>()
if (original) {
@ -81,7 +81,10 @@ export const stage = Effect.fn("SessionRevert.stage")(function* (input: {
const revert = {
messageID: input.messageID,
snapshot: original,
diff: files.map((file) => file.patch).join("").trim(),
diff: files
.map((file) => file.patch)
.join("")
.trim(),
files,
} satisfies SessionSchema.Info["revert"]
yield* events.publish(SessionEvent.RevertEvent.Staged, {

31
packages/core/src/session/runner/llm.ts
View file

@ -309,19 +309,24 @@ export const layer = Layer.effect(
const stepSettlement = publisher.stepSettlement()
if (stepSettlement && !publisher.hasProviderError()) {
const endSnapshot = yield* snapshots.capture()
const files = startSnapshot && endSnapshot
? yield* snapshots.files({ from: startSnapshot, to: endSnapshot }).pipe(Effect.catch(() => Effect.succeed(undefined)))
: undefined
yield* withPublication(events.publish(SessionEvent.Step.Ended, {
sessionID: session.id,
timestamp: yield* DateTime.now,
assistantMessageID: yield* publisher.startAssistant(),
finish: stepSettlement.finish,
cost: 0,
tokens: stepSettlement.tokens,
snapshot: endSnapshot,
files,
}))
const files =
startSnapshot && endSnapshot
? yield* snapshots
.files({ from: startSnapshot, to: endSnapshot })
.pipe(Effect.catch(() => Effect.succeed(undefined)))
: undefined
yield* withPublication(
events.publish(SessionEvent.Step.Ended, {
sessionID: session.id,
timestamp: yield* DateTime.now,
assistantMessageID: yield* publisher.startAssistant(),
finish: stepSettlement.finish,
cost: 0,
tokens: stepSettlement.tokens,
snapshot: endSnapshot,
files,
}),
)
}
if (publisher.hasProviderError())
yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted"))

30
packages/core/src/snapshot.ts
View file

@ -111,11 +111,13 @@ export const layer = Layer.effect(
gitDirectory,
commonDirectory: gitDirectory,
})
return yield* git.repo.create({
worktree,
gitDirectory,
seed: source,
}).pipe(Effect.mapError((cause) => failure("capture", cause)))
return yield* git.repo
.create({
worktree,
gitDirectory,
seed: source,
})
.pipe(Effect.mapError((cause) => failure("capture", cause)))
})
const enabled = Effect.fnUntraced(function* () {
@ -136,9 +138,7 @@ export const layer = Layer.effect(
}),
)
}).pipe(
Effect.catch((cause) =>
Effect.logWarning("failed to capture snapshot", { cause }).pipe(Effect.as(undefined)),
),
Effect.catch((cause) => Effect.logWarning("failed to capture snapshot", { cause }).pipe(Effect.as(undefined))),
)
})
@ -189,12 +189,14 @@ export const layer = Layer.effect(
if (!(yield* enabled())) return yield* new Error({ operation: "preview", message: "Snapshots are disabled" })
const repo = yield* repository().pipe(Effect.mapError((cause) => failure("preview", cause)))
const files = yield* plan("preview", input)
const current = yield* git.tree.capture({
repository: repo,
scopes: Array.from(files.keys()),
ignores: source,
maximumUntrackedFileBytes: 2 * 1024 * 1024,
}).pipe(Effect.mapError((cause) => failure("preview", cause)))
const current = yield* git.tree
.capture({
repository: repo,
scopes: Array.from(files.keys()),
ignores: source,
maximumUntrackedFileBytes: 2 * 1024 * 1024,
})
.pipe(Effect.mapError((cause) => failure("preview", cause)))
return yield* git.tree
.preview({
repository: repo,

50
packages/core/test/session-projector.test.ts
View file

@ -46,18 +46,52 @@ describe("SessionProjector", () => {
it.effect("projects staged, cleared, and committed reverts", () =>
Effect.gen(function* () {
const db = (yield* Database.Service).db
yield* db.insert(ProjectTable).values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }).run()
yield* db.insert(SessionTable).values({ id: sessionID, project_id: Project.ID.global, slug: "test", directory: "/project", title: "test", version: "test" }).run()
yield* db
.insert(ProjectTable)
.values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
.run()
yield* db
.insert(SessionTable)
.values({
id: sessionID,
project_id: Project.ID.global,
slug: "test",
directory: "/project",
title: "test",
version: "test",
})
.run()
const boundary = SessionMessage.ID.make("msg_boundary")
yield* db.insert(SessionMessageTable).values([assistantRow(boundary, 1), assistantRow(SessionMessage.ID.make("msg_later"), 2)]).run()
yield* db
.insert(SessionMessageTable)
.values([assistantRow(boundary, 1), assistantRow(SessionMessage.ID.make("msg_later"), 2)])
.run()
const events = yield* EventV2.Service
yield* events.publish(SessionEvent.RevertEvent.Staged, { sessionID, timestamp: DateTime.makeUnsafe(1), revert: { messageID: boundary, snapshot: Snapshot.ID.make("tree"), diff: "patch", files: [] } })
expect((yield* db.select({ revert: SessionTable.revert }).from(SessionTable).get())?.revert).toMatchObject({ messageID: boundary, snapshot: "tree", files: [] })
yield* events.publish(SessionEvent.RevertEvent.Staged, {
sessionID,
timestamp: DateTime.makeUnsafe(1),
revert: { messageID: boundary, snapshot: Snapshot.ID.make("tree"), diff: "patch", files: [] },
})
expect((yield* db.select({ revert: SessionTable.revert }).from(SessionTable).get())?.revert).toMatchObject({
messageID: boundary,
snapshot: "tree",
files: [],
})
yield* events.publish(SessionEvent.RevertEvent.Cleared, { sessionID, timestamp: DateTime.makeUnsafe(2) })
expect((yield* db.select({ revert: SessionTable.revert }).from(SessionTable).get())?.revert).toBeNull()
yield* events.publish(SessionEvent.RevertEvent.Staged, { sessionID, timestamp: DateTime.makeUnsafe(3), revert: { messageID: boundary, files: [] } })
yield* events.publish(SessionEvent.RevertEvent.Committed, { sessionID, messageID: boundary, timestamp: DateTime.makeUnsafe(4) })
expect((yield* db.select({ id: SessionMessageTable.id }).from(SessionMessageTable).all()).map((row) => row.id)).toEqual([boundary])
yield* events.publish(SessionEvent.RevertEvent.Staged, {
sessionID,
timestamp: DateTime.makeUnsafe(3),
revert: { messageID: boundary, files: [] },
})
yield* events.publish(SessionEvent.RevertEvent.Committed, {
sessionID,
messageID: boundary,
timestamp: DateTime.makeUnsafe(4),
})
expect(
(yield* db.select({ id: SessionMessageTable.id }).from(SessionMessageTable).all()).map((row) => row.id),
).toEqual([boundary])
}),
)

8
packages/core/test/snapshot.test.ts
View file

@ -123,8 +123,12 @@ describe("Snapshot", () => {
),
),
)
expect(yield* Effect.promise(() => fs.stat(path.join(tmp.path, "snapshot", projectID, Hash.fast(project))))).toBeDefined()
expect(yield* Effect.promise(() => fs.stat(path.join(tmp.path, "snapshot", projectID, Hash.fast(linked))))).toBeDefined()
expect(
yield* Effect.promise(() => fs.stat(path.join(tmp.path, "snapshot", projectID, Hash.fast(project)))),
).toBeDefined()
expect(
yield* Effect.promise(() => fs.stat(path.join(tmp.path, "snapshot", projectID, Hash.fast(linked)))),
).toBeDefined()
}),
(tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
),

1008
packages/sdk/openapi.json
View file

File diff suppressed because it is too large Load diff

6
packages/server/src/handlers/message.ts
View file

@ -58,7 +58,11 @@ export const MessageHandler = HttpApiBuilder.group(Api, "server.message", (handl
const ref = `err_${crypto.randomUUID().slice(0, 8)}`
return Effect.logError("failed to decode session message").pipe(
Effect.annotateLogs({ ref, sessionID: error.sessionID, messageID: error.messageID }),
Effect.andThen(Effect.fail(new UnknownError({ message: "Unexpected server error. Check server logs for details.", ref }))),
Effect.andThen(
Effect.fail(
new UnknownError({ message: "Unexpected server error. Check server logs for details.", ref }),
),
),
)
}),
)

6
packages/server/src/handlers/session.ts
View file

@ -307,7 +307,11 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl
const ref = `err_${crypto.randomUUID().slice(0, 8)}`
return Effect.logError("failed to decode session message").pipe(
Effect.annotateLogs({ ref, sessionID: error.sessionID, messageID: error.messageID }),
Effect.andThen(Effect.fail(new UnknownError({ message: "Unexpected server error. Check server logs for details.", ref }))),
Effect.andThen(
Effect.fail(
new UnknownError({ message: "Unexpected server error. Check server logs for details.", ref }),
),
),
)
}),
),