chore(observability): merge v2

This commit is contained in:
starptech 2026-07-09 06:36:36 +02:00
commit f4e661f874
276 changed files with 17001 additions and 3755 deletions

View file

@ -1,6 +1,6 @@
{
"$schema": "https://json.schemastore.org/package.json",
"version": "1.17.14",
"version": "1.17.15",
"name": "@opencode-ai/core",
"type": "module",
"license": "MIT",

View file

@ -1,8 +1,10 @@
{
"version": "7",
"dialect": "sqlite",
"id": "b0355fd9-bf41-42e3-9dca-76107de27ecd",
"prevIds": ["95328a41-789d-44de-9643-6ac6ecd6b4ec"],
"id": "beb91b32-23e2-405f-99b8-1e8763db94f8",
"prevIds": [
"00000000-0000-0000-0000-000000000000"
],
"ddl": [
{
"name": "workspace",
@ -1016,11 +1018,11 @@
},
{
"type": "text",
"notNull": false,
"notNull": true,
"autoincrement": false,
"default": null,
"generated": null,
"name": "prompt",
"name": "data",
"entityType": "columns",
"table": "session_input"
},
@ -1575,9 +1577,13 @@
"table": "session_share"
},
{
"columns": ["project_id"],
"columns": [
"project_id"
],
"tableTo": "project",
"columnsTo": ["id"],
"columnsTo": [
"id"
],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@ -1586,9 +1592,13 @@
"table": "workspace"
},
{
"columns": ["active_account_id"],
"columns": [
"active_account_id"
],
"tableTo": "account",
"columnsTo": ["id"],
"columnsTo": [
"id"
],
"onUpdate": "NO ACTION",
"onDelete": "SET NULL",
"nameExplicit": false,
@ -1597,9 +1607,13 @@
"table": "account_state"
},
{
"columns": ["aggregate_id"],
"columns": [
"aggregate_id"
],
"tableTo": "event_sequence",
"columnsTo": ["aggregate_id"],
"columnsTo": [
"aggregate_id"
],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@ -1608,9 +1622,13 @@
"table": "event"
},
{
"columns": ["project_id"],
"columns": [
"project_id"
],
"tableTo": "project",
"columnsTo": ["id"],
"columnsTo": [
"id"
],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@ -1619,9 +1637,13 @@
"table": "permission"
},
{
"columns": ["project_id"],
"columns": [
"project_id"
],
"tableTo": "project",
"columnsTo": ["id"],
"columnsTo": [
"id"
],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@ -1630,9 +1652,13 @@
"table": "project_directory"
},
{
"columns": ["session_id"],
"columns": [
"session_id"
],
"tableTo": "session",
"columnsTo": ["id"],
"columnsTo": [
"id"
],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@ -1641,9 +1667,13 @@
"table": "instruction_checkpoint"
},
{
"columns": ["session_id"],
"columns": [
"session_id"
],
"tableTo": "session",
"columnsTo": ["id"],
"columnsTo": [
"id"
],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@ -1652,9 +1682,13 @@
"table": "instruction_entry"
},
{
"columns": ["session_id"],
"columns": [
"session_id"
],
"tableTo": "session",
"columnsTo": ["id"],
"columnsTo": [
"id"
],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@ -1663,9 +1697,13 @@
"table": "message"
},
{
"columns": ["message_id"],
"columns": [
"message_id"
],
"tableTo": "message",
"columnsTo": ["id"],
"columnsTo": [
"id"
],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@ -1674,9 +1712,13 @@
"table": "part"
},
{
"columns": ["session_id"],
"columns": [
"session_id"
],
"tableTo": "session",
"columnsTo": ["id"],
"columnsTo": [
"id"
],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@ -1685,9 +1727,13 @@
"table": "session_input"
},
{
"columns": ["session_id"],
"columns": [
"session_id"
],
"tableTo": "session",
"columnsTo": ["id"],
"columnsTo": [
"id"
],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@ -1696,9 +1742,13 @@
"table": "session_message"
},
{
"columns": ["project_id"],
"columns": [
"project_id"
],
"tableTo": "project",
"columnsTo": ["id"],
"columnsTo": [
"id"
],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@ -1707,9 +1757,13 @@
"table": "session"
},
{
"columns": ["session_id"],
"columns": [
"session_id"
],
"tableTo": "session",
"columnsTo": ["id"],
"columnsTo": [
"id"
],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@ -1718,9 +1772,13 @@
"table": "todo"
},
{
"columns": ["session_id"],
"columns": [
"session_id"
],
"tableTo": "session",
"columnsTo": ["id"],
"columnsTo": [
"id"
],
"onUpdate": "NO ACTION",
"onDelete": "CASCADE",
"nameExplicit": false,
@ -1729,140 +1787,184 @@
"table": "session_share"
},
{
"columns": ["email", "url"],
"columns": [
"email",
"url"
],
"nameExplicit": false,
"name": "control_account_pk",
"entityType": "pks",
"table": "control_account"
},
{
"columns": ["project_id", "directory"],
"columns": [
"project_id",
"directory"
],
"nameExplicit": false,
"name": "project_directory_pk",
"entityType": "pks",
"table": "project_directory"
},
{
"columns": ["session_id", "key"],
"columns": [
"session_id",
"key"
],
"nameExplicit": false,
"name": "instruction_entry_pk",
"entityType": "pks",
"table": "instruction_entry"
},
{
"columns": ["session_id", "position"],
"columns": [
"session_id",
"position"
],
"nameExplicit": false,
"name": "todo_pk",
"entityType": "pks",
"table": "todo"
},
{
"columns": ["id"],
"columns": [
"id"
],
"nameExplicit": false,
"name": "workspace_pk",
"table": "workspace",
"entityType": "pks"
},
{
"columns": ["name"],
"columns": [
"name"
],
"nameExplicit": false,
"name": "data_migration_pk",
"table": "data_migration",
"entityType": "pks"
},
{
"columns": ["id"],
"columns": [
"id"
],
"nameExplicit": false,
"name": "account_state_pk",
"table": "account_state",
"entityType": "pks"
},
{
"columns": ["id"],
"columns": [
"id"
],
"nameExplicit": false,
"name": "account_pk",
"table": "account",
"entityType": "pks"
},
{
"columns": ["id"],
"columns": [
"id"
],
"nameExplicit": false,
"name": "credential_pk",
"table": "credential",
"entityType": "pks"
},
{
"columns": ["aggregate_id"],
"columns": [
"aggregate_id"
],
"nameExplicit": false,
"name": "event_sequence_pk",
"table": "event_sequence",
"entityType": "pks"
},
{
"columns": ["id"],
"columns": [
"id"
],
"nameExplicit": false,
"name": "event_pk",
"table": "event",
"entityType": "pks"
},
{
"columns": ["id"],
"columns": [
"id"
],
"nameExplicit": false,
"name": "permission_pk",
"table": "permission",
"entityType": "pks"
},
{
"columns": ["id"],
"columns": [
"id"
],
"nameExplicit": false,
"name": "project_pk",
"table": "project",
"entityType": "pks"
},
{
"columns": ["session_id"],
"columns": [
"session_id"
],
"nameExplicit": false,
"name": "instruction_checkpoint_pk",
"table": "instruction_checkpoint",
"entityType": "pks"
},
{
"columns": ["id"],
"columns": [
"id"
],
"nameExplicit": false,
"name": "message_pk",
"table": "message",
"entityType": "pks"
},
{
"columns": ["id"],
"columns": [
"id"
],
"nameExplicit": false,
"name": "part_pk",
"table": "part",
"entityType": "pks"
},
{
"columns": ["id"],
"columns": [
"id"
],
"nameExplicit": false,
"name": "session_input_pk",
"table": "session_input",
"entityType": "pks"
},
{
"columns": ["id"],
"columns": [
"id"
],
"nameExplicit": false,
"name": "session_message_pk",
"table": "session_message",
"entityType": "pks"
},
{
"columns": ["id"],
"columns": [
"id"
],
"nameExplicit": false,
"name": "session_pk",
"table": "session",
"entityType": "pks"
},
{
"columns": ["session_id"],
"columns": [
"session_id"
],
"nameExplicit": false,
"name": "session_share_pk",
"table": "session_share",
@ -1994,10 +2096,6 @@
"value": "promoted_seq",
"isExpression": false
},
{
"value": "type",
"isExpression": false
},
{
"value": "delivery",
"isExpression": false
@ -2010,7 +2108,7 @@
"isUnique": false,
"where": null,
"origin": "manual",
"name": "session_input_session_pending_type_delivery_seq_idx",
"name": "session_input_session_pending_delivery_seq_idx",
"entityType": "indexes",
"table": "session_input"
},
@ -2198,4 +2296,4 @@
}
],
"renames": []
}
}

View file

@ -119,7 +119,17 @@ export class Directory extends Schema.Class<Directory>("Config.Directory")({
path: AbsolutePath,
}) {}
export type Entry = Document | Directory
export class AgentsDirectory extends Schema.Class<AgentsDirectory>("Config.AgentsDirectory")({
type: Schema.Literal("agents"),
path: AbsolutePath,
}) {}
export class ClaudeDirectory extends Schema.Class<ClaudeDirectory>("Config.ClaudeDirectory")({
type: Schema.Literal("claude"),
path: AbsolutePath,
}) {}
export type Entry = Document | Directory | AgentsDirectory | ClaudeDirectory
export function latest<K extends keyof Info>(entries: readonly Entry[], key: K): Info[K] | undefined {
return entries
@ -176,16 +186,37 @@ const layer = Layer.effect(
const discover = Effect.fn("Config.discover")(function* () {
const globalDirectory = AbsolutePath.make(global.config)
const globalAgentsDirectory = AbsolutePath.make(path.join(global.home, ".agents"))
const globalClaudeDirectory = AbsolutePath.make(path.join(global.home, ".claude"))
const locationIsGlobal = path.resolve(location.directory) === path.resolve(global.config)
const discovered = locationIsGlobal
? []
: yield* fs
.up({
targets: [".opencode", ...names.toReversed()],
targets: [".opencode", ".claude", ".agents", ...names.toReversed()],
start: location.directory,
stop: location.project.directory,
})
.pipe(Effect.orDie)
.pipe(Effect.orDie)
// We load certain files from a few other folders in the ecosystem
const claude = [
...((yield* fs.isDir(globalClaudeDirectory))
? [new ClaudeDirectory({ type: "claude", path: globalClaudeDirectory })]
: []),
...discovered
.filter((item) => path.basename(item) === ".claude")
.map((directory) => new ClaudeDirectory({ type: "claude", path: AbsolutePath.make(directory) })),
]
const agents = [
...((yield* fs.isDir(globalAgentsDirectory))
? [new AgentsDirectory({ type: "agents", path: globalAgentsDirectory })]
: []),
...discovered
.filter((item) => path.basename(item) === ".agents")
.map((directory) => new AgentsDirectory({ type: "agents", path: AbsolutePath.make(directory) })),
]
const directories = [
globalDirectory,
...discovered
@ -193,15 +224,18 @@ const layer = Layer.effect(
.toReversed()
.map((directory) => AbsolutePath.make(directory)),
]
const directPaths = discovered.filter((item) => path.basename(item) !== ".opencode").toReversed()
const directPaths = discovered
.filter((item) => ![".agents", ".claude", ".opencode"].includes(path.basename(item)))
.toReversed()
const direct = yield* Effect.forEach(directPaths, loadFile).pipe(
Effect.orDie,
Effect.map((configs) => configs.filter((config): config is Document => config !== undefined)),
)
const supplementary = yield* Effect.forEach(directories, loadDirectory).pipe(Effect.orDie)
return {
entries: [...(supplementary[0] ?? []), ...direct, ...supplementary.slice(1).flat()],
directories,
entries: [...claude, ...agents, ...(supplementary[0] ?? []), ...direct, ...supplementary.slice(1).flat()],
directories: [...directories, ...claude.map((entry) => entry.path), ...agents.map((entry) => entry.path)],
files: directPaths,
}
})

View file

@ -27,12 +27,10 @@ export const Selection = Schema.Union([Short, Explicit])
.annotate({ identifier: "Config.ModelSelection" })
function parse(input: string): Selection {
const providerEnd = input.indexOf("/")
const variantStart = input.lastIndexOf("#")
const hasVariant = variantStart > providerEnd
const ref = Model.Ref.parse(input)
return {
providerID: Provider.ID.make(input.slice(0, providerEnd)),
model: Model.ID.make(input.slice(providerEnd + 1, hasVariant ? variantStart : undefined)),
...(hasVariant ? { variant: Model.VariantID.make(input.slice(variantStart + 1)) } : {}),
providerID: ref.providerID,
model: ref.id,
...(ref.variant ? { variant: ref.variant } : {}),
}
}

View file

@ -10,6 +10,11 @@ import { ConfigMarkdown } from "../markdown"
import { FSUtil } from "../../fs-util"
import { ConfigAgentV1 } from "../../v1/config/agent"
import { ConfigMigrateV1 } from "../../v1/config/migrate"
import { Global } from "../../global"
import { PermissionV2 } from "../../permission"
import type { LocationMutation } from "../../location-mutation"
import type { ReadTool } from "../../tool/read"
import type { EditTool } from "../../tool/edit"
const legacySources = [
{ pattern: "{agent,agents}/**/*.md", primary: false },
@ -18,6 +23,11 @@ const legacySources = [
const decodeAgent = Schema.decodeUnknownOption(ConfigAgent.Info)
const decodeLegacyAgent = Schema.decodeUnknownOption(ConfigAgentV1.Info)
const decodeConfig = Schema.decodeUnknownOption(Config.Info)
type PathAction =
| LocationMutation.ExternalDirectoryAuthorization["action"]
| typeof ReadTool.name
| typeof EditTool.name
const pathActions = ["external_directory", "read", "edit"] as const satisfies readonly PathAction[]
const agentKeys = new Set([
"model",
"variant",
@ -37,9 +47,11 @@ export const Plugin = define({
effect: Effect.fn(function* (ctx) {
const config = yield* Config.Service
const fs = yield* FSUtil.Service
const global = yield* Global.Service
const load = Effect.fn("ConfigAgentPlugin.load")(function* () {
return yield* Effect.forEach(yield* config.entries(), (entry) => {
if (entry.type === "document") return Effect.succeed([entry])
if (entry.type !== "directory") return Effect.succeed([])
return Effect.gen(function* () {
const files = yield* discover(fs, entry.path)
return yield* Effect.forEach(files, (file) =>
@ -57,11 +69,14 @@ export const Plugin = define({
})
const loaded = { documents: yield* load() }
yield* ctx.agent.transform((draft) => {
const global = loaded.documents.flatMap((document) => document.info.permissions ?? [])
const permissions = expandPermissions(
loaded.documents.flatMap((document) => document.info.permissions ?? []),
global.home,
)
const configuredDefault = Config.latest(loaded.documents, "default_agent")
if (configuredDefault !== undefined) draft.default(AgentV2.ID.make(configuredDefault))
for (const current of draft.list()) {
draft.update(current.id, (agent) => agent.permissions.push(...global))
draft.update(current.id, (agent) => agent.permissions.push(...permissions))
}
for (const document of loaded.documents) {
@ -74,7 +89,7 @@ export const Plugin = define({
const exists = draft.get(agentID) !== undefined
draft.update(agentID, (agent) => {
if (!exists) agent.permissions.push(...global)
if (!exists) agent.permissions.push(...permissions)
if (item.model !== undefined)
agent.model = {
id: item.model.model,
@ -91,7 +106,9 @@ export const Plugin = define({
if (item.hidden !== undefined) agent.hidden = item.hidden
if (item.color !== undefined) agent.color = item.color
if (item.steps !== undefined) agent.steps = item.steps
if (item.permissions !== undefined) agent.permissions.push(...item.permissions)
if (item.permissions !== undefined) {
agent.permissions.push(...expandPermissions(item.permissions, global.home))
}
})
}
}
@ -109,6 +126,27 @@ export const Plugin = define({
}),
})
function expandPermissions(rules: PermissionV2.Ruleset, home: string): PermissionV2.Ruleset {
// Expand only resources tools resolve as filesystem paths. Bash resources are raw shell text:
// rewriting `$HOME/private/**` would miss `$HOME/private/key`, and safe expansion needs shell-aware parsing.
return rules.map((rule) =>
isPathAction(rule.action) ? { ...rule, resource: expandHome(rule.resource, home) } : rule,
)
}
function isPathAction(action: string): action is PathAction {
return pathActions.some((item) => item === action)
}
function expandHome(resource: string, home: string) {
if (resource.startsWith("~/")) return home + resource.slice(1)
if (resource === "~") return home
if (resource === "$HOME") return home
if (resource.startsWith("$HOME/")) return home + resource.slice(5)
if (resource.startsWith("$HOME\\")) return home + resource.slice(5)
return resource
}
function discover(fs: FSUtil.Interface, directory: string) {
return Effect.forEach(legacySources, (source) =>
fs

View file

@ -19,6 +19,7 @@ export const Plugin = define({
const load = Effect.fn("ConfigCommandPlugin.load")(function* () {
return yield* Effect.forEach(yield* config.entries(), (entry) => {
if (entry.type === "document") return Effect.succeed([{ commands: entry.info.commands }])
if (entry.type !== "directory") return Effect.succeed([])
return loadDirectory(fs, entry.path).pipe(
Effect.map((commands) => [
{ commands: Object.fromEntries(commands.map((command) => [command.name, command.info])) },

View file

@ -17,8 +17,18 @@ export const Plugin = define({
const location = yield* Location.Service
const loaded = { entries: yield* config.entries() }
yield* ctx.skill.transform((draft) => {
const claude = loaded.entries.flatMap((entry) => (entry.type === "claude" ? [entry.path] : []))
const agents = loaded.entries.flatMap((entry) => (entry.type === "agents" ? [entry.path] : []))
const directories = loaded.entries.flatMap((entry) => (entry.type === "directory" ? [entry.path] : []))
const items = loaded.entries.flatMap((entry) => (entry.type === "document" ? (entry.info.skills ?? []) : []))
for (const directory of [...claude, ...agents]) {
draft.source(
SkillV2.DirectorySource.make({
type: "directory",
path: AbsolutePath.make(path.join(directory, "skills")),
}),
)
}
for (const directory of directories) {
draft.source(
SkillV2.DirectorySource.make({ type: "directory", path: AbsolutePath.make(path.join(directory, "skill")) }),

View file

@ -49,5 +49,6 @@ export const migrations = (
import("./migration/20260706223930_add-session-fork"),
import("./migration/20260707010146_durable_session_inbox"),
import("./migration/20260707120000_migrate_prelaunch_v2_state"),
import("./migration/20260709013000_generic_session_input"),
])
).map((module) => module.default) satisfies DatabaseMigration.Migration[]

View file

@ -0,0 +1,77 @@
import { Effect } from "effect"
import type { DatabaseMigration } from "../migration"
export default {
id: "20260709013000_generic_session_input",
up(tx) {
return Effect.gen(function* () {
yield* tx.run(`PRAGMA foreign_keys=OFF;`)
yield* tx.run(`
DELETE FROM \`event\`
WHERE \`type\` IN ('session.prompt.admitted.1', 'session.prompt.promoted.1')
AND json_extract(\`data\`, '$.inputID') IN (
SELECT \`id\` FROM \`session_input\` WHERE \`type\` = 'prompt' AND \`prompt\` IS NULL
);
`)
yield* tx.run(`
CREATE TABLE \`__new_session_input\` (
\`id\` text PRIMARY KEY,
\`session_id\` text NOT NULL,
\`type\` text NOT NULL,
\`data\` text NOT NULL,
\`delivery\` text,
\`admitted_seq\` integer NOT NULL,
\`promoted_seq\` integer,
\`time_created\` integer NOT NULL,
CONSTRAINT \`fk_session_input_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
INSERT INTO \`__new_session_input\`(
\`id\`, \`session_id\`, \`type\`, \`data\`, \`delivery\`, \`admitted_seq\`, \`promoted_seq\`, \`time_created\`
)
SELECT
\`id\`, \`session_id\`, CASE WHEN \`type\` = 'prompt' THEN 'user' ELSE \`type\` END,
CASE WHEN \`type\` = 'prompt' THEN \`prompt\` ELSE '{}' END,
\`delivery\`, \`admitted_seq\`, \`promoted_seq\`, \`time_created\`
FROM \`session_input\`
WHERE \`type\` != 'prompt' OR \`prompt\` IS NOT NULL;
`)
yield* tx.run(`DROP TABLE \`session_input\`;`)
yield* tx.run(`ALTER TABLE \`__new_session_input\` RENAME TO \`session_input\`;`)
yield* tx.run(`PRAGMA foreign_keys=ON;`)
yield* tx.run(
`CREATE INDEX \`session_input_session_pending_delivery_seq_idx\` ON \`session_input\` (\`session_id\`,\`promoted_seq\`,\`delivery\`,\`admitted_seq\`);`,
)
yield* tx.run(
`CREATE UNIQUE INDEX \`session_input_session_pending_compaction_idx\` ON \`session_input\` (\`session_id\`) WHERE \`type\` = 'compaction' and \`promoted_seq\` is null;`,
)
yield* tx.run(
`CREATE UNIQUE INDEX \`session_input_session_admitted_seq_idx\` ON \`session_input\` (\`session_id\`,\`admitted_seq\`);`,
)
yield* tx.run(
`CREATE UNIQUE INDEX \`session_input_session_promoted_seq_idx\` ON \`session_input\` (\`session_id\`,\`promoted_seq\`);`,
)
yield* tx.run(`
UPDATE \`event\`
SET
\`type\` = 'session.input.admitted.1',
\`data\` = json_object(
'sessionID', json_extract(\`data\`, '$.sessionID'),
'inputID', json_extract(\`data\`, '$.inputID'),
'input', json_object(
'type', 'user',
'data', json_extract(\`data\`, '$.prompt'),
'delivery', json_extract(\`data\`, '$.delivery')
)
)
WHERE \`type\` = 'session.prompt.admitted.1';
`)
yield* tx.run(`
UPDATE \`event\`
SET \`type\` = 'session.input.promoted.1'
WHERE \`type\` = 'session.prompt.promoted.1';
`)
})
},
} satisfies DatabaseMigration.Migration

View file

@ -171,7 +171,7 @@ export default {
\`id\` text PRIMARY KEY,
\`session_id\` text NOT NULL,
\`type\` text NOT NULL,
\`prompt\` text,
\`data\` text NOT NULL,
\`delivery\` text,
\`admitted_seq\` integer NOT NULL,
\`promoted_seq\` integer,
@ -262,7 +262,7 @@ export default {
yield* tx.run(`CREATE INDEX \`part_message_id_id_idx\` ON \`part\` (\`message_id\`,\`id\`);`)
yield* tx.run(`CREATE INDEX \`part_session_idx\` ON \`part\` (\`session_id\`);`)
yield* tx.run(
`CREATE INDEX \`session_input_session_pending_type_delivery_seq_idx\` ON \`session_input\` (\`session_id\`,\`promoted_seq\`,\`type\`,\`delivery\`,\`admitted_seq\`);`,
`CREATE INDEX \`session_input_session_pending_delivery_seq_idx\` ON \`session_input\` (\`session_id\`,\`promoted_seq\`,\`delivery\`,\`admitted_seq\`);`,
)
yield* tx.run(
`CREATE UNIQUE INDEX \`session_input_session_pending_compaction_idx\` ON \`session_input\` (\`session_id\`) WHERE "session_input"."type" = 'compaction' and "session_input"."promoted_seq" is null;`,

View file

@ -62,9 +62,10 @@ export namespace FSUtil {
})
const readFileStringSafe = Effect.fn("FileSystem.readFileStringSafe")(function* (path: string) {
return yield* fs
.readFileString(path)
.pipe(Effect.catchReason("PlatformError", "NotFound", () => Effect.succeed(undefined)))
return yield* fs.readFileString(path).pipe(
Effect.catchReason("PlatformError", "NotFound", () => Effect.succeed(undefined)),
Effect.catchReason("PlatformError", "PermissionDenied", () => Effect.succeed(undefined)),
)
})
const isDir = Effect.fn("FileSystem.isDir")(function* (path: string) {

View file

@ -203,6 +203,7 @@ type AttemptTime = { created: number; expires: number }
type PendingAttempt = {
status: "pending"
completing: boolean
persisting: boolean
authorization: OAuthAuthorization
integrationID: ID
methodID: MethodID
@ -320,27 +321,61 @@ const layer = Layer.effect(
}
const settle = Effect.fnUntraced(function* (attemptID: AttemptID, exit: Exit.Exit<Credential.OAuth, unknown>) {
const now = yield* Clock.currentTimeMillis
const result = yield* SynchronizedRef.modify(attempts, (current) => {
const attempt = current.get(attemptID)
if (!attempt || attempt.status !== "pending") return [undefined, current]
const terminal: TerminalAttempt = Exit.isSuccess(exit)
? { status: "complete", time: attempt.time, removeAt: now + terminalRetention }
: { status: "failed", message: message(exit.cause), time: attempt.time, removeAt: now + terminalRetention }
return [attempt, new Map(current).set(attemptID, terminal)]
})
if (!result) return
if (Exit.isSuccess(exit)) {
const implementation = state.get().integrations.get(result.integrationID)?.implementations.get(result.methodID)
yield* credentials.create({
integrationID: result.integrationID,
label: result.label ?? implementation?.label?.(exit.value),
value: exit.value,
})
yield* events.publish(Event.ConnectionUpdated, { integrationID: result.integrationID })
yield* events.publish(Event.Updated, {})
}
yield* close(result.scope)
return yield* Effect.uninterruptible(
Effect.gen(function* () {
const now = yield* Clock.currentTimeMillis
const attempt = yield* SynchronizedRef.modify(attempts, (current) => {
const match = current.get(attemptID)
if (!match || match.status !== "pending" || match.persisting) return [undefined, current]
const next = Exit.isSuccess(exit)
? { ...match, persisting: true }
: {
status: "failed" as const,
message: message(exit.cause),
time: match.time,
removeAt: now + terminalRetention,
}
return [match, new Map(current).set(attemptID, next)]
})
if (!attempt) return
if (Exit.isFailure(exit)) {
yield* close(attempt.scope)
return
}
yield* Effect.gen(function* () {
const implementation = state
.get()
.integrations.get(attempt.integrationID)
?.implementations.get(attempt.methodID)
const persistence = yield* Effect.sync(() => attempt.label ?? implementation?.label?.(exit.value)).pipe(
Effect.flatMap((label) =>
credentials.create({
integrationID: attempt.integrationID,
label,
value: exit.value,
}),
),
Effect.asVoid,
Effect.exit,
)
const settledAt = yield* Clock.currentTimeMillis
const terminal: TerminalAttempt = Exit.isSuccess(persistence)
? { status: "complete", time: attempt.time, removeAt: settledAt + terminalRetention }
: {
status: "failed",
message: message(persistence.cause),
time: attempt.time,
removeAt: settledAt + terminalRetention,
}
// Persisting attempts cannot be cancelled, expired, or claimed again.
yield* SynchronizedRef.update(attempts, (current) => new Map(current).set(attemptID, terminal))
if (Exit.isFailure(persistence)) yield* Effect.failCause(persistence.cause)
yield* events.publish(Event.ConnectionUpdated, { integrationID: attempt.integrationID })
yield* events.publish(Event.Updated, {})
}).pipe(Effect.ensuring(close(attempt.scope)))
}),
)
})
const scrub = Effect.fnUntraced(function* () {
@ -349,7 +384,7 @@ const layer = Layer.effect(
const next = new Map(current)
const scopes: Scope.Closeable[] = []
for (const [id, attempt] of current) {
if (attempt.status === "pending" && attempt.time.expires <= now) {
if (attempt.status === "pending" && !attempt.persisting && attempt.time.expires <= now) {
scopes.push(attempt.scope)
next.set(id, { status: "expired", time: attempt.time, removeAt: now + terminalRetention })
continue
@ -432,6 +467,7 @@ const layer = Layer.effect(
new Map(current).set(id, {
status: "pending",
completing: authorization.mode === "auto",
persisting: false,
authorization,
integrationID: input.integrationID,
methodID: input.methodID,
@ -506,7 +542,7 @@ const layer = Layer.effect(
cancel: Effect.fn("Integration.attempt.cancel")(function* (attemptID) {
const attempt = yield* SynchronizedRef.modify(attempts, (current) => {
const match = current.get(attemptID)
if (!match || match.status !== "pending") return [undefined, current]
if (!match || match.status !== "pending" || match.persisting) return [undefined, current]
const next = new Map(current)
next.delete(attemptID)
return [match, next]

View file

@ -43,14 +43,21 @@ function oauth(http: HttpClient.HttpClient) {
type: "oauth",
label: "OpenCode Console account",
},
authorize: () =>
authorize: (inputs) =>
Effect.gen(function* () {
const device = yield* post(http, `${defaultServer}/auth/device/code`, { client_id: clientID }, Device)
const server = yield* normalizeServer(inputs.server ?? defaultServer)
const device = yield* post(http, `${server}/auth/device/code`, { client_id: clientID }, Device)
const verification = URL.canParse(device.verification_uri_complete)
? new URL(device.verification_uri_complete)
: undefined
if (verification && verification.protocol !== "http:" && verification.protocol !== "https:") {
return yield* Effect.fail(new Error("Invalid device verification URL: expected HTTP(S)"))
}
return {
mode: "auto" as const,
url: `${defaultServer}${device.verification_uri_complete}`,
url: verification?.href ?? `${server}/${device.verification_uri_complete.replace(/^\/+/, "")}`,
instructions: `Enter code: ${device.user_code}`,
callback: poll(http, defaultServer, device.device_code, Duration.seconds(device.interval)),
callback: poll(http, server, device.device_code, Duration.seconds(device.interval)),
}
}),
refresh: (credential) =>
@ -219,6 +226,18 @@ function withoutCredentials(body: Readonly<Record<string, unknown>> | undefined)
return Object.fromEntries(Object.entries(body ?? {}).filter(([key]) => key !== "apiKey" && key !== "headers"))
}
function normalizeServer(input: string) {
return Effect.try({
try: () => {
const url = new URL(input)
if (url.protocol !== "http:" && url.protocol !== "https:") throw new Error("expected HTTP(S)")
return `${url.origin}${url.pathname.replace(/\/+$/, "")}`
},
catch: (cause) =>
new Error(`Invalid OpenCode server URL: ${cause instanceof Error ? cause.message : String(cause)}`),
})
}
function remoteCost(input: NonNullable<(typeof ConfigProviderV1.Model.Type)["cost"]>) {
const base = {
input: Money.USDPerMillionTokens.make(input.input),

View file

@ -92,6 +92,7 @@ const configuredPlugins = Effect.fn("SkillPlugin.configuredPlugins")(function* (
}),
)
}
if (entry.type !== "directory") return Effect.succeed([])
return fs
.glob("{plugin,plugins}/*.{ts,js}", {
cwd: entry.path,

View file

@ -122,6 +122,13 @@ export class PromptConflictError extends Schema.TaggedErrorClass<PromptConflictE
sessionID: SessionSchema.ID,
messageID: SessionMessage.ID,
}) {}
export class SyntheticConflictError extends Schema.TaggedErrorClass<SyntheticConflictError>()(
"Session.SyntheticConflictError",
{
sessionID: SessionSchema.ID,
inputID: SessionMessage.ID,
},
) {}
export class AttachmentError extends Schema.TaggedErrorClass<AttachmentError>()("Session.AttachmentError", {
uri: Schema.String,
message: Schema.String,
@ -147,6 +154,7 @@ export type Error =
| MessageDecodeError
| OperationUnavailableError
| PromptConflictError
| SyntheticConflictError
| AttachmentError
| CompactionConflictError
| BusyError
@ -204,10 +212,13 @@ export interface Interface {
readonly prompt: (input: {
id?: SessionMessage.ID
sessionID: SessionSchema.ID
prompt: PromptInput.Prompt
text: string
files?: PromptInput.Prompt["files"]
agents?: PromptInput.Prompt["agents"]
metadata?: Record<string, unknown>
delivery?: SessionInput.Delivery
resume?: boolean
}) => Effect.Effect<SessionInput.Admitted, NotFoundError | PromptConflictError | AttachmentError>
}) => Effect.Effect<SessionInput.User, NotFoundError | PromptConflictError | AttachmentError>
readonly command: (input: {
id?: SessionMessage.ID
sessionID: SessionSchema.ID
@ -220,7 +231,7 @@ export interface Interface {
delivery?: SessionInput.Delivery
resume?: boolean
}) => Effect.Effect<
SessionInput.Admitted,
SessionInput.User,
NotFoundError | PromptConflictError | AttachmentError | CommandV2.NotFoundError | CommandV2.EvaluationError
>
readonly shell: (input: {
@ -243,12 +254,14 @@ export interface Interface {
readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, NotFoundError | SessionRunner.RunError>
readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect<void>
readonly synthetic: (input: {
id?: SessionMessage.ID
sessionID: SessionSchema.ID
text: string
description?: string
metadata?: Record<string, unknown>
delivery?: SessionInput.Delivery
resume?: boolean
}) => Effect.Effect<void, NotFoundError>
}) => Effect.Effect<SessionInput.Synthetic, NotFoundError | SyntheticConflictError>
readonly revert: {
readonly stage: (input: {
sessionID: SessionSchema.ID
@ -487,15 +500,19 @@ const layer = Layer.effect(
// continues from the reverted boundary rather than stale post-boundary history.
if (session.revert)
yield* SessionRevert.commit(session).pipe(Effect.provideService(EventV2.Service, events))
const prompt = yield* resolvePrompt(input.prompt).pipe(Effect.provideService(FSUtil.Service, fs))
const prompt = yield* resolvePrompt({ text: input.text, files: input.files, agents: input.agents }).pipe(
Effect.provideService(FSUtil.Service, fs),
)
const messageID = input.id ?? SessionMessage.ID.create()
const delivery = input.delivery ?? "steer"
const expected = { sessionID: input.sessionID, messageID, prompt, delivery }
const admittedInput = SessionInput.Message.make({
type: "user",
data: { ...prompt, metadata: input.metadata },
delivery: input.delivery ?? "steer",
})
const admitted = yield* SessionInput.admit(db, events, {
id: messageID,
sessionID: input.sessionID,
prompt,
delivery,
input: admittedInput,
}).pipe(
Effect.catchDefect((defect) =>
defect instanceof SessionInput.LifecycleConflict
@ -503,7 +520,10 @@ const layer = Layer.effect(
: Effect.die(defect),
),
)
if (!SessionInput.equivalent(admitted, expected))
if (
admitted.type !== "user" ||
!SessionInput.equivalent(admitted, { sessionID: input.sessionID, input: admittedInput })
)
return yield* new PromptConflictError({ sessionID: input.sessionID, messageID })
if (input.resume !== false) {
if (activeShells.has(admitted.sessionID)) return admitted
@ -550,7 +570,9 @@ const layer = Layer.effect(
return yield* result.prompt({
id: input.id,
sessionID: input.sessionID,
prompt: { text: evaluated.text, files: input.files, agents: input.agents },
text: evaluated.text,
files: input.files,
agents: input.agents,
delivery: input.delivery,
resume: input.resume,
})
@ -675,35 +697,60 @@ const layer = Layer.effect(
yield* result.get(sessionID)
const backgrounded = yield* jobs.backgroundAll({ sessionID })
if (backgrounded.length === 0) return
yield* result.synthetic({
sessionID,
text: [
"User requested that active blocking work be moved to the background.",
"",
"Backgrounded work:",
...backgrounded.map((job) => `- ${job.type}: ${job.title && job.title.length > 0 ? job.title : job.id}`),
"",
"The backgrounded work is still unfinished. Move on to other work if you can. If there is nothing else useful to do, finish your response. Do not wait, sleep, poll, or report the backgrounded work as complete until a later completion notification is added to the conversation.",
].join("\n"),
})
yield* result
.synthetic({
sessionID,
text: [
"User requested that active blocking work be moved to the background.",
"",
"Backgrounded work:",
...backgrounded.map((job) => `- ${job.type}: ${job.title && job.title.length > 0 ? job.title : job.id}`),
"",
"The backgrounded work is still unfinished. Move on to other work if you can. If there is nothing else useful to do, finish your response. Do not wait, sleep, poll, or report the backgrounded work as complete until a later completion notification is added to the conversation.",
].join("\n"),
})
.pipe(Effect.catchTag("Session.SyntheticConflictError", Effect.die))
}),
resume: Effect.fn("V2Session.resume")(function* (sessionID) {
yield* result.get(sessionID)
yield* execution.resume(sessionID)
}),
synthetic: Effect.fn("V2Session.synthetic")(function* (input) {
yield* result.get(input.sessionID)
yield* events.publish(SessionEvent.Synthetic, {
sessionID: input.sessionID,
text: input.text,
description: input.description,
metadata: input.metadata,
})
if (input.resume === false) return
yield* execution
.resume(input.sessionID)
.pipe(Effect.ignore, Effect.forkIn(scope, { startImmediately: true }), Effect.asVoid)
}),
synthetic: Effect.fn("V2Session.synthetic")((input) =>
Effect.uninterruptible(
Effect.gen(function* () {
yield* result.get(input.sessionID)
const inputID = input.id ?? SessionMessage.ID.create()
const admittedInput = SessionInput.Message.make({
type: "synthetic",
data: {
text: input.text,
description: input.description,
metadata: input.metadata,
},
delivery: input.delivery ?? "steer",
})
const admitted = yield* SessionInput.admit(db, events, {
id: inputID,
sessionID: input.sessionID,
input: admittedInput,
}).pipe(
Effect.catchDefect((defect) =>
defect instanceof SessionInput.LifecycleConflict
? new SyntheticConflictError({ sessionID: input.sessionID, inputID })
: Effect.die(defect),
),
)
if (
admitted.type !== "synthetic" ||
!SessionInput.equivalent(admitted, { sessionID: input.sessionID, input: admittedInput })
)
return yield* new SyntheticConflictError({ sessionID: input.sessionID, inputID })
if (input.resume !== false && !(yield* result.get(input.sessionID)).revert)
yield* execution.wake(input.sessionID)
return admitted
}),
),
),
interrupt: Effect.fn("V2Session.interrupt")((sessionID) =>
Effect.uninterruptible(execution.interrupt(sessionID)),
),
@ -721,10 +768,12 @@ const layer = Layer.effect(
clear: Effect.fn("V2Session.revert.clear")(function* (sessionID) {
const session = yield* result.get(sessionID)
if ((yield* execution.active).has(sessionID)) return yield* new BusyError({ sessionID })
return yield* SessionRevert.clear(session).pipe(
const revert = yield* SessionRevert.clear(session).pipe(
Effect.provideService(EventV2.Service, events),
Effect.provide(locations.get(session.location)),
)
yield* execution.wake(sessionID)
return revert
}),
commit: Effect.fn("V2Session.revert.commit")(function* (sessionID) {
const session = yield* result.get(sessionID)

View file

@ -2,22 +2,32 @@ export * as SessionInput from "./input"
import { and, asc, eq, isNull } from "drizzle-orm"
import { DateTime, Effect, Schema } from "effect"
import { Admitted, Compaction, Delivery, Info, PromptEntry } from "@opencode-ai/schema/session-input"
import {
Compaction,
Delivery,
Info,
Message,
Synthetic,
SyntheticData,
User,
UserData,
} from "@opencode-ai/schema/session-input"
import type { Database } from "../database/database"
import type { EventV2 } from "../event"
import { KeyedMutex } from "../effect/keyed-mutex"
import { SessionEvent } from "./event"
import { SessionMessage } from "./message"
import { Prompt } from "@opencode-ai/schema/prompt"
import { SessionSchema } from "./schema"
import { SessionInputTable, SessionMessageTable } from "./sql"
type DatabaseService = Database.Interface["db"]
export { Admitted, Compaction, Delivery, Info, PromptEntry }
export { Compaction, Delivery, Info, Message, Synthetic, SyntheticData, User, UserData }
const decodePrompt = Schema.decodeUnknownSync(Prompt)
const encodePrompt = Schema.encodeSync(Prompt)
const decodeUser = Schema.decodeUnknownSync(UserData)
const encodeUser = Schema.encodeSync(UserData)
const decodeSynthetic = Schema.decodeUnknownSync(SyntheticData)
const encodeSynthetic = Schema.encodeSync(SyntheticData)
const inboxLocks = KeyedMutex.makeUnsafe<SessionSchema.ID>()
export class LifecycleConflict extends Schema.TaggedErrorClass<LifecycleConflict>()("SessionInput.LifecycleConflict", {
@ -37,27 +47,26 @@ const fromRow = (row: typeof SessionInputTable.$inferSelect): Info => {
type: "compaction",
...(row.promoted_seq === null ? {} : { handledSeq: row.promoted_seq }),
})
if (!row.prompt || !row.delivery) throw new LifecycleConflict({ id: base.id })
return PromptEntry.make({
...base,
type: "prompt",
prompt: decodePrompt(row.prompt),
delivery: row.delivery,
...(row.promoted_seq === null ? {} : { promotedSeq: row.promoted_seq }),
})
if (!row.delivery) throw new LifecycleConflict({ id: base.id })
if (row.type === "user")
return User.make({
...base,
type: "user",
data: decodeUser(row.data),
delivery: row.delivery,
...(row.promoted_seq === null ? {} : { promotedSeq: row.promoted_seq }),
})
if (row.type === "synthetic")
return Synthetic.make({
...base,
type: "synthetic",
data: decodeSynthetic(row.data),
delivery: row.delivery,
...(row.promoted_seq === null ? {} : { promotedSeq: row.promoted_seq }),
})
throw new LifecycleConflict({ id: base.id })
}
const toAdmitted = (entry: PromptEntry): Admitted =>
Admitted.make({
admittedSeq: entry.admittedSeq,
id: entry.id,
sessionID: entry.sessionID,
prompt: entry.prompt,
delivery: entry.delivery,
timeCreated: entry.timeCreated,
...(entry.promotedSeq === undefined ? {} : { promotedSeq: entry.promotedSeq }),
})
export const find = Effect.fn("SessionInput.find")(function* (db: DatabaseService, id: SessionMessage.ID) {
const row = yield* db.select().from(SessionInputTable).where(eq(SessionInputTable.id, id)).get().pipe(Effect.orDie)
return row === undefined ? undefined : fromRow(row)
@ -89,44 +98,43 @@ export const pendingCompaction = Effect.fn("SessionInput.pendingCompaction")(fun
export const admit = Effect.fn("SessionInput.admit")(function* (
db: DatabaseService,
events: EventV2.Interface,
input: {
request: {
readonly id: SessionMessage.ID
readonly sessionID: SessionSchema.ID
readonly prompt: Prompt
readonly delivery: Delivery
readonly input: Message
},
) {
const existing = yield* find(db, input.id)
const existing = yield* find(db, request.id)
if (existing !== undefined) {
if (existing.type !== "prompt") return yield* Effect.die(new LifecycleConflict({ id: input.id }))
return toAdmitted(existing)
if (existing.type === "compaction") return yield* Effect.die(new LifecycleConflict({ id: request.id }))
return existing
}
return yield* events
.publish(SessionEvent.PromptAdmitted, {
inputID: input.id,
sessionID: input.sessionID,
prompt: input.prompt,
delivery: input.delivery,
.publish(SessionEvent.InputAdmitted, {
inputID: request.id,
sessionID: request.sessionID,
input: request.input,
})
.pipe(
Effect.flatMap((event) =>
event.durable === undefined
? Effect.die(new Error("Prompt admission event is missing aggregate sequence"))
: Effect.succeed(
Admitted.make({
admittedSeq: event.durable.seq,
id: input.id,
sessionID: input.sessionID,
prompt: input.prompt,
delivery: input.delivery,
timeCreated: event.created,
}),
),
),
Effect.flatMap((event) => {
if (event.durable === undefined)
return Effect.die(new Error("Session input admission event is missing aggregate sequence"))
const base = {
admittedSeq: event.durable.seq,
id: request.id,
sessionID: request.sessionID,
timeCreated: event.created,
}
return Effect.succeed(
request.input.type === "user"
? User.make({ ...base, ...request.input })
: Synthetic.make({ ...base, ...request.input }),
)
}),
Effect.catchDefect((defect) =>
find(db, input.id).pipe(
find(db, request.id).pipe(
Effect.flatMap((stored) =>
stored?.type === "prompt" ? Effect.succeed(toAdmitted(stored)) : Effect.die(defect),
stored?.type === request.input.type ? Effect.succeed(stored) : Effect.die(defect),
),
),
),
@ -174,38 +182,37 @@ export const admitCompaction = Effect.fn("SessionInput.admitCompaction")(functio
export const projectAdmitted = Effect.fn("SessionInput.projectAdmitted")(function* (
db: DatabaseService,
input: {
request: {
readonly admittedSeq: number
readonly id: SessionMessage.ID
readonly sessionID: SessionSchema.ID
readonly prompt: Prompt
readonly delivery: Delivery
readonly input: Message
readonly timeCreated: DateTime.Utc
},
) {
const message = yield* db
.select({ id: SessionMessageTable.id })
.from(SessionMessageTable)
.where(eq(SessionMessageTable.id, input.id))
.where(eq(SessionMessageTable.id, request.id))
.get()
.pipe(Effect.orDie)
if (message !== undefined) return yield* Effect.die(new LifecycleConflict({ id: input.id }))
if (message !== undefined) return yield* Effect.die(new LifecycleConflict({ id: request.id }))
const stored = yield* db
.insert(SessionInputTable)
.values({
id: input.id,
session_id: input.sessionID,
type: "prompt",
admitted_seq: input.admittedSeq,
prompt: encodePrompt(input.prompt),
delivery: input.delivery,
time_created: DateTime.toEpochMillis(input.timeCreated),
id: request.id,
session_id: request.sessionID,
type: request.input.type,
data: request.input.type === "user" ? encodeUser(request.input.data) : encodeSynthetic(request.input.data),
delivery: request.input.delivery,
admitted_seq: request.admittedSeq,
time_created: DateTime.toEpochMillis(request.timeCreated),
})
.onConflictDoNothing()
.returning({ id: SessionInputTable.id })
.get()
.pipe(Effect.orDie)
if (!stored) return yield* Effect.die(new LifecycleConflict({ id: input.id }))
if (!stored) return yield* Effect.die(new LifecycleConflict({ id: request.id }))
})
export const projectCompactionAdmitted = Effect.fn("SessionInput.projectCompactionAdmitted")(function* (
@ -230,6 +237,7 @@ export const projectCompactionAdmitted = Effect.fn("SessionInput.projectCompacti
id: input.id,
session_id: input.sessionID,
type: "compaction",
data: {},
admitted_seq: input.admittedSeq,
time_created: DateTime.toEpochMillis(input.timeCreated),
})
@ -246,7 +254,7 @@ export const projectCompactionAdmitted = Effect.fn("SessionInput.projectCompacti
return yield* Effect.die(new LifecycleConflict({ id: input.id }))
})
export const projectPromptPromoted = Effect.fn("SessionInput.projectPromptPromoted")(function* (
export const projectPromoted = Effect.fn("SessionInput.projectPromoted")(function* (
db: DatabaseService,
input: {
readonly id: SessionMessage.ID
@ -262,23 +270,16 @@ export const projectPromptPromoted = Effect.fn("SessionInput.projectPromptPromot
and(
eq(SessionInputTable.id, input.id),
eq(SessionInputTable.session_id, input.sessionID),
eq(SessionInputTable.type, "prompt"),
isNull(SessionInputTable.promoted_seq),
),
)
.returning()
.get()
.pipe(Effect.orDie)
if (updated) {
const stored = fromRow(updated)
if (stored.type !== "prompt" || stored.sessionID !== input.sessionID)
return yield* Effect.die(new LifecycleConflict({ id: input.id }))
return stored
}
const stored = yield* find(db, input.id)
const stored = updated ? fromRow(updated) : yield* find(db, input.id)
if (
!stored ||
stored.type !== "prompt" ||
stored.type === "compaction" ||
stored.sessionID !== input.sessionID ||
stored.promotedSeq !== input.promotedSeq
)
@ -322,7 +323,6 @@ export const hasPending = Effect.fn("SessionInput.hasPending")(function* (
.where(
and(
eq(SessionInputTable.session_id, sessionID),
eq(SessionInputTable.type, "prompt"),
isNull(SessionInputTable.promoted_seq),
eq(SessionInputTable.delivery, delivery),
),
@ -334,16 +334,21 @@ export const hasPending = Effect.fn("SessionInput.hasPending")(function* (
})
export const equivalent = (
input: Admitted,
expected: {
readonly sessionID: SessionSchema.ID
readonly prompt: Prompt
readonly delivery: Delivery
},
) =>
input.delivery === expected.delivery &&
input.sessionID === expected.sessionID &&
JSON.stringify(encodePrompt(input.prompt)) === JSON.stringify(encodePrompt(expected.prompt))
input: User | Synthetic,
expected: { readonly sessionID: SessionSchema.ID; readonly input: Message },
) => {
if (
input.type !== expected.input.type ||
input.delivery !== expected.input.delivery ||
input.sessionID !== expected.sessionID
)
return false
if (input.type === "user" && expected.input.type === "user")
return JSON.stringify(encodeUser(input.data)) === JSON.stringify(encodeUser(expected.input.data))
if (input.type === "synthetic" && expected.input.type === "synthetic")
return JSON.stringify(encodeSynthetic(input.data)) === JSON.stringify(encodeSynthetic(expected.input.data))
return false
}
const publish = Effect.fn("SessionInput.publish")(function* (
db: DatabaseService,
@ -358,9 +363,9 @@ const publish = Effect.fn("SessionInput.publish")(function* (
rows,
(row) => {
const entry = fromRow(row)
if (entry.type !== "prompt") return Effect.die(new LifecycleConflict({ id: entry.id }))
if (entry.type === "compaction") return Effect.die(new LifecycleConflict({ id: entry.id }))
return events
.publish(SessionEvent.PromptPromoted, {
.publish(SessionEvent.InputPromoted, {
sessionID,
inputID: entry.id,
})
@ -369,7 +374,7 @@ const publish = Effect.fn("SessionInput.publish")(function* (
defect instanceof LifecycleConflict
? find(db, entry.id).pipe(
Effect.flatMap((stored) =>
stored?.type === "prompt" && stored.promotedSeq !== undefined
stored?.type !== "compaction" && stored?.promotedSeq !== undefined
? Effect.void
: Effect.die(defect),
),
@ -397,7 +402,6 @@ export const promoteSteers = Effect.fn("SessionInput.promoteSteers")(function* (
.where(
and(
eq(SessionInputTable.session_id, sessionID),
eq(SessionInputTable.type, "prompt"),
isNull(SessionInputTable.promoted_seq),
eq(SessionInputTable.delivery, "steer"),
),
@ -420,7 +424,6 @@ export const promoteNextQueued = Effect.fn("SessionInput.promoteNextQueued")(fun
.where(
and(
eq(SessionInputTable.session_id, sessionID),
eq(SessionInputTable.type, "prompt"),
isNull(SessionInputTable.promoted_seq),
eq(SessionInputTable.delivery, "queue"),
),

View file

@ -172,8 +172,8 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
"session.renamed": () => Effect.void,
"session.deleted": () => Effect.void,
"session.forked": () => Effect.void,
"session.prompt.promoted": () => Effect.void,
"session.prompt.admitted": () => Effect.void,
"session.input.promoted": () => Effect.void,
"session.input.admitted": () => Effect.void,
"session.execution.started": () => Effect.void,
"session.execution.succeeded": () => clearCurrentRetry,
"session.execution.failed": () => clearCurrentRetry,

View file

@ -313,13 +313,13 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
.values(
inputRows.flatMap((row) => {
const id = idMap.get(row.id)
return id && row.type === "prompt"
return id && row.type !== "compaction"
? [
{
id,
session_id: event.data.sessionID,
type: "prompt" as const,
prompt: row.prompt,
type: row.type,
data: row.data,
delivery: row.delivery,
admitted_seq: row.admitted_seq,
promoted_seq: row.promoted_seq,
@ -629,27 +629,40 @@ const layer = Layer.effectDiscard(
.pipe(Effect.orDie),
)
yield* events.project(SessionEvent.Forked, (event) => projectFork(db, event))
yield* events.project(SessionEvent.PromptPromoted, (event) =>
yield* events.project(SessionEvent.InputPromoted, (event) =>
Effect.gen(function* () {
if (event.durable === undefined)
return yield* Effect.die(new Error("Durable Session event is missing aggregate sequence"))
const input = yield* SessionInput.projectPromptPromoted(db, {
const input = yield* SessionInput.projectPromoted(db, {
id: event.data.inputID,
sessionID: event.data.sessionID,
promotedSeq: event.durable.seq,
})
yield* insertMessage(db, event, {
id: input.id,
type: "user",
metadata: event.metadata,
text: input.prompt.text,
files: input.prompt.files,
agents: input.prompt.agents,
time: { created: event.created },
})
yield* insertMessage(
db,
event,
input.type === "user"
? {
id: input.id,
type: "user",
metadata: input.data.metadata,
text: input.data.text,
files: input.data.files,
agents: input.data.agents,
time: { created: event.created },
}
: {
id: input.id,
type: "synthetic",
text: input.data.text,
description: input.data.description,
metadata: input.data.metadata,
time: { created: event.created },
},
)
}),
)
yield* events.project(SessionEvent.PromptAdmitted, (event) =>
yield* events.project(SessionEvent.InputAdmitted, (event) =>
Effect.gen(function* () {
if (event.durable === undefined)
return yield* Effect.die(new Error("Durable Session event is missing aggregate sequence"))
@ -657,8 +670,7 @@ const layer = Layer.effectDiscard(
admittedSeq: event.durable.seq,
id: event.data.inputID,
sessionID: event.data.sessionID,
prompt: event.data.prompt,
delivery: event.data.delivery,
input: event.data.input,
timeCreated: event.created,
})
}),

View file

@ -3,7 +3,6 @@ import { sql } from "drizzle-orm"
import { directoryColumn, pathColumn } from "../database/path"
import { ProjectTable } from "../project/sql"
import type { SessionMessage } from "./message"
import type { Prompt } from "@opencode-ai/schema/prompt"
import type { SessionInput } from "./input"
import type { FileDiff } from "@opencode-ai/schema/file-diff"
import { PermissionV1 } from "../v1/permission"
@ -14,6 +13,7 @@ import { WorkspaceV2 } from "../workspace"
import { Timestamps } from "../database/schema.sql"
import type { Instructions } from "../instructions/index"
import type { Session } from "@opencode-ai/schema/session"
import type { SyntheticData, UserData } from "@opencode-ai/schema/session-input"
import type { RevertV1 } from "@opencode-ai/schema/session-revert"
import type { Schema } from "effect"
@ -150,7 +150,7 @@ export const SessionInputTable = sqliteTable(
.notNull()
.references(() => SessionTable.id, { onDelete: "cascade" }),
type: text().$type<SessionInput.Info["type"]>().notNull(),
prompt: text({ mode: "json" }).$type<Prompt>(),
data: text({ mode: "json" }).$type<UserData | SyntheticData | Record<string, never>>().notNull(),
delivery: text().$type<SessionInput.Delivery>(),
admitted_seq: integer().notNull(),
promoted_seq: integer(),
@ -159,10 +159,9 @@ export const SessionInputTable = sqliteTable(
.$default(() => Date.now()),
},
(table) => [
index("session_input_session_pending_type_delivery_seq_idx").on(
index("session_input_session_pending_delivery_seq_idx").on(
table.session_id,
table.promoted_seq,
table.type,
table.delivery,
table.admitted_seq,
),

View file

@ -81,6 +81,7 @@ export const Plugin = {
forms
.ask({
sessionID: context.sessionID,
title: "Questions",
metadata: {
kind: "question",
tool: { messageID: context.assistantMessageID, callID: context.toolCallID },

View file

@ -154,7 +154,7 @@ export const Plugin = {
const run = Effect.gen(function* () {
// The child session owns its agent/model (set at create); prompt only admits input.
yield* runtime.session.prompt({ sessionID: child.id, prompt: { text: input.prompt }, resume: false })
yield* runtime.session.prompt({ sessionID: child.id, text: input.prompt, resume: false })
yield* telemetry.resume(runtime.session.resume(child.id), background)
return yield* latestAssistantText(child.id)
}).pipe(Effect.onInterrupt(() => runtime.session.interrupt(child.id)))

View file

@ -8,13 +8,15 @@ import { ConfigAgentPlugin } from "@opencode-ai/core/config/plugin/agent"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
import { LayerNode } from "@opencode-ai/core/effect/layer-node"
import { FSUtil } from "@opencode-ai/core/fs-util"
import { Global } from "@opencode-ai/core/global"
import { PermissionV2 } from "@opencode-ai/core/permission"
import { AbsolutePath } from "@opencode-ai/core/schema"
import { ConfigMigrateV1 } from "@opencode-ai/core/v1/config/migrate"
import { tmpdir } from "../fixture/tmpdir"
import { testEffect } from "../lib/effect"
import { agentHost, host } from "../plugin/host"
const it = testEffect(AppNodeBuilder.build(LayerNode.group([AgentV2.node, FSUtil.node])))
const it = testEffect(AppNodeBuilder.build(LayerNode.group([AgentV2.node, FSUtil.node, Global.node])))
const decode = Schema.decodeUnknownSync(Config.Info)
const defaultPermissions = [
{ action: "*", resource: "*", effect: "allow" },
@ -22,6 +24,33 @@ const defaultPermissions = [
] satisfies PermissionV2.Ruleset
describe("ConfigAgentPlugin.Plugin", () => {
it.effect("matches POSIX paths against home-relative permissions", () =>
Effect.gen(function* () {
const permissions = yield* loadHomePermissions("/home/test")
expect(PermissionV2.evaluate("external_directory", "/home/test/p/opencode/src/*", permissions).effect).toBe(
"allow",
)
expect(PermissionV2.evaluate("external_directory", "/home/test/cache/files/*", permissions).effect).toBe("deny")
expect(PermissionV2.evaluate("external_directory", "/some/~/path", permissions).effect).toBe("deny")
expect(PermissionV2.evaluate("external_directory", "$HOMELESS/private/*", permissions).effect).toBe("deny")
expect(permissions).toContainEqual({ action: "shell", resource: "$HOME/private/**", effect: "deny" })
expect(permissions).not.toContainEqual({ action: "shell", resource: "/home/test/private/**", effect: "deny" })
expect(PermissionV2.evaluate("shell", "$HOME/private/key", permissions).effect).toBe("deny")
}),
)
it.effect("matches Windows paths against home-relative permissions", () =>
Effect.gen(function* () {
const permissions = yield* loadHomePermissions("C:\\Users\\test")
expect(
PermissionV2.evaluate("external_directory", "C:\\Users\\test\\p\\opencode\\src\\*", permissions).effect,
).toBe("allow")
expect(PermissionV2.evaluate("external_directory", "C:\\Users\\test\\cache\\files\\*", permissions).effect).toBe(
"deny",
)
}),
)
it.effect("applies all global permissions before agent-specific permissions", () =>
Effect.gen(function* () {
const agents = yield* AgentV2.Service
@ -280,3 +309,51 @@ Use native v2 fields.`,
),
)
})
function loadHomePermissions(home: string) {
return Effect.gen(function* () {
const agents = yield* AgentV2.Service
const build = AgentV2.ID.make("build")
yield* agents.transform((editor) => editor.update(build, () => {}))
const config = Config.Service.of({
entries: () =>
Effect.succeed([
new Config.Document({
type: "document",
info: decode(
ConfigMigrateV1.migrate({
permission: {
external_directory: {
"~/p/**": "allow",
"/some/~/path": "deny",
"$HOMELESS/**": "deny",
},
bash: {
"$HOME/private/**": "deny",
},
},
agent: {
build: {
permission: {
external_directory: {
"$HOME/cache/**": "deny",
},
},
},
},
}),
),
}),
]),
})
yield* ConfigAgentPlugin.Plugin.effect(host({ agent: agentHost(agents) })).pipe(
Effect.provideService(Config.Service, config),
Effect.provideService(Global.Service, Global.Service.of({ ...Global.make(), home })),
)
const agent = yield* agents.get(build)
if (!agent) throw new Error("expected configured build agent")
return agent.permissions
})
}

View file

@ -44,7 +44,7 @@ function testLayer(
)
return AppNodeBuilder.build(LayerNode.group([Config.node, EventV2.node]), [
[Location.node, locationLayer],
[Global.node, Global.layerWith({ config: globalDirectory })],
[Global.node, Global.layerWith({ config: globalDirectory, home: path.join(globalDirectory, "home") })],
...(watcher ? ([[Watcher.node, watcher]] as const) : []),
])
}
@ -112,6 +112,7 @@ describe("Config", () => {
info: new Config.Info({ model: selection("openrouter/openai/gpt-5") }),
}),
new Config.Directory({ type: "directory", path: AbsolutePath.make("/skills") }),
new Config.AgentsDirectory({ type: "agents", path: AbsolutePath.make("/agents") }),
new Config.Document({ type: "document", info: new Config.Info({}) }),
new Config.Document({
type: "document",
@ -848,11 +849,19 @@ describe("Config", () => {
const root = path.join(tmp.path, "repo")
const parent = path.join(root, "packages")
const directory = path.join(parent, "app")
const globalAgents = path.join(global, "home", ".agents")
const globalClaude = path.join(global, "home", ".claude")
return Effect.gen(function* () {
yield* Effect.promise(async () => {
await fs.mkdir(global, { recursive: true })
await fs.mkdir(globalAgents, { recursive: true })
await fs.mkdir(globalClaude, { recursive: true })
await fs.mkdir(directory, { recursive: true })
await fs.mkdir(path.join(root, ".agents"), { recursive: true })
await fs.mkdir(path.join(root, ".claude"), { recursive: true })
await fs.mkdir(path.join(root, ".opencode"), { recursive: true })
await fs.mkdir(path.join(directory, ".agents"), { recursive: true })
await fs.mkdir(path.join(directory, ".claude"), { recursive: true })
await fs.mkdir(path.join(directory, ".opencode"), { recursive: true })
await Promise.all([
fs.writeFile(path.join(tmp.path, "opencode.json"), JSON.stringify({ $schema: "outside" })),
@ -878,6 +887,16 @@ describe("Config", () => {
AbsolutePath.make(path.join(root, ".opencode")),
AbsolutePath.make(path.join(directory, ".opencode")),
])
expect(entries.filter((entry) => entry.type === "agents").map((entry) => entry.path)).toEqual([
AbsolutePath.make(globalAgents),
AbsolutePath.make(path.join(directory, ".agents")),
AbsolutePath.make(path.join(root, ".agents")),
])
expect(entries.filter((entry) => entry.type === "claude").map((entry) => entry.path)).toEqual([
AbsolutePath.make(globalClaude),
AbsolutePath.make(path.join(directory, ".claude")),
AbsolutePath.make(path.join(root, ".claude")),
])
expect(documents.map((document) => document.info.$schema)).toEqual([
"global",
"root",
@ -887,6 +906,12 @@ describe("Config", () => {
"directory-dot",
])
expect(entries.map((entry) => (entry.type === "document" ? entry.info.$schema : entry.path))).toEqual([
AbsolutePath.make(globalClaude),
AbsolutePath.make(path.join(directory, ".claude")),
AbsolutePath.make(path.join(root, ".claude")),
AbsolutePath.make(globalAgents),
AbsolutePath.make(path.join(directory, ".agents")),
AbsolutePath.make(path.join(root, ".agents")),
"global",
AbsolutePath.make(global),
"root",

View file

@ -46,6 +46,8 @@ describe("ConfigSkillPlugin.Plugin", () => {
Config.Service.of({
entries: () =>
Effect.succeed([
new Config.ClaudeDirectory({ type: "claude", path: AbsolutePath.make("/repo/.claude") }),
new Config.AgentsDirectory({ type: "agents", path: AbsolutePath.make("/repo/.agents") }),
new Config.Directory({ type: "directory", path: AbsolutePath.make("/repo/.opencode") }),
new Config.Document({
type: "document",
@ -59,6 +61,14 @@ describe("ConfigSkillPlugin.Plugin", () => {
)
expect(sources).toEqual([
SkillV2.DirectorySource.make({
type: "directory",
path: AbsolutePath.make(path.join("/repo/.claude", "skills")),
}),
SkillV2.DirectorySource.make({
type: "directory",
path: AbsolutePath.make(path.join("/repo/.agents", "skills")),
}),
SkillV2.DirectorySource.make({
type: "directory",
path: AbsolutePath.make(path.join("/repo/.opencode", "skill")),

View file

@ -18,6 +18,7 @@ import simplifySessionInputMigration from "@opencode-ai/core/database/migration/
import resetSessionEventsMigration from "@opencode-ai/core/database/migration/20260703200000_reset_v2_session_events"
import durableSessionInboxMigration from "@opencode-ai/core/database/migration/20260707010146_durable_session_inbox"
import migratePrelaunchV2StateMigration from "@opencode-ai/core/database/migration/20260707120000_migrate_prelaunch_v2_state"
import genericSessionInputMigration from "@opencode-ai/core/database/migration/20260709013000_generic_session_input"
import renameInstructionsMigration from "@opencode-ai/core/database/migration/20260705180000_rename_instructions"
import addSessionForkMigration from "@opencode-ai/core/database/migration/20260706223930_add-session-fork"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
@ -369,7 +370,7 @@ describe("DatabaseMigration", () => {
{ name: "event_aggregate_type_seq_idx" },
{ name: "session_input_session_admitted_seq_idx" },
{ name: "session_input_session_pending_compaction_idx" },
{ name: "session_input_session_pending_type_delivery_seq_idx" },
{ name: "session_input_session_pending_delivery_seq_idx" },
{ name: "session_input_session_promoted_seq_idx" },
{ name: "session_message_session_seq_idx" },
{ name: "session_message_session_time_created_id_idx" },
@ -607,7 +608,7 @@ describe("DatabaseMigration", () => {
sql`INSERT INTO event (id, aggregate_id, seq, type, data, created) VALUES ('event', 'session', 9, 'session.updated.1', '{}', 1)`,
)
yield* db.run(
sql`INSERT INTO session_input (id, session_id, type, prompt, delivery, admitted_seq, time_created) VALUES ('input', 'session', 'prompt', '{}', 'steer', 9, 1)`,
sql`INSERT INTO session_input (id, session_id, type, data, delivery, admitted_seq, time_created) VALUES ('input', 'session', 'user', '{}', 'steer', 9, 1)`,
)
yield* db.run(
sql`INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data) VALUES ('projected', 'session', 'user', 9, 1, 1, '{}')`,
@ -701,6 +702,56 @@ describe("DatabaseMigration", () => {
)
})
test("migrates prompt inbox rows and lifecycle events to generic user input", async () => {
await run(
Effect.gen(function* () {
const db = yield* makeDb
yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY)`)
yield* db.run(sql`INSERT INTO session (id) VALUES ('session')`)
yield* db.run(
sql`CREATE TABLE session_input (id text PRIMARY KEY, session_id text NOT NULL, type text NOT NULL, prompt text, delivery text, admitted_seq integer NOT NULL, promoted_seq integer, time_created integer NOT NULL)`,
)
yield* db.run(
sql`INSERT INTO session_input (id, session_id, type, prompt, delivery, admitted_seq, promoted_seq, time_created) VALUES ('input', 'session', 'prompt', '{"text":"hello"}', 'queue', 4, NULL, 1)`,
)
yield* db.run(
sql`INSERT INTO session_input (id, session_id, type, prompt, delivery, admitted_seq, promoted_seq, time_created) VALUES ('empty', 'session', 'prompt', NULL, 'steer', 6, NULL, 2)`,
)
yield* db.run(
sql`CREATE TABLE event (id text PRIMARY KEY, aggregate_id text NOT NULL, seq integer NOT NULL, created integer NOT NULL, type text NOT NULL, data text NOT NULL)`,
)
yield* db.run(
sql`INSERT INTO event (id, aggregate_id, seq, created, type, data) VALUES ('admitted', 'session', 4, 1, 'session.prompt.admitted.1', '{"sessionID":"session","inputID":"input","prompt":{"text":"hello"},"delivery":"queue"}')`,
)
yield* db.run(
sql`INSERT INTO event (id, aggregate_id, seq, created, type, data) VALUES ('promoted', 'session', 5, 2, 'session.prompt.promoted.1', '{"sessionID":"session","inputID":"input"}')`,
)
yield* db.run(
sql`INSERT INTO event (id, aggregate_id, seq, created, type, data) VALUES ('empty-admitted', 'session', 6, 2, 'session.prompt.admitted.1', '{"sessionID":"session","inputID":"empty","prompt":null,"delivery":"steer"}')`,
)
yield* db.run(
sql`INSERT INTO event (id, aggregate_id, seq, created, type, data) VALUES ('empty-promoted', 'session', 7, 2, 'session.prompt.promoted.1', '{"sessionID":"session","inputID":"empty"}')`,
)
yield* DatabaseMigration.applyOnly(db, [genericSessionInputMigration])
expect(yield* db.all(sql`SELECT id, type, data, delivery FROM session_input ORDER BY admitted_seq`)).toEqual([
{ id: "input", type: "user", data: '{"text":"hello"}', delivery: "queue" },
])
expect(yield* db.all(sql`SELECT type, data FROM event ORDER BY seq`)).toEqual([
{
type: "session.input.admitted.1",
data: '{"sessionID":"session","inputID":"input","input":{"type":"user","data":{"text":"hello"},"delivery":"queue"}}',
},
{
type: "session.input.promoted.1",
data: '{"sessionID":"session","inputID":"input"}',
},
])
}),
)
})
test("resets incompatible projected Session messages before adding sequence order", async () => {
await run(
Effect.gen(function* () {

View file

@ -185,35 +185,33 @@ describeWatcher("LocationWatcher", () => {
),
)
it.live("watches non-git roots", () =>
it.live("skips non-git roots", () =>
withTmp((directory) =>
Effect.gen(function* () {
const fs = yield* FSUtil.Service
const file = path.join(directory, "plain.txt")
yield* ready(directory)
expect(yield* nextUpdate((event) => event.file === file, fs.writeFileString(file, "plain"))).toEqual({
file,
event: "add",
})
yield* noUpdate((event) => event.file === file, fs.writeFileString(file, "plain"))
}),
),
)
it.live("ignores dependency, VCS, and build directories at any depth", () =>
withTmp((directory) =>
Effect.gen(function* () {
const afs = yield* FSUtil.Service
yield* ready(directory)
const roots = ["node_modules", ".git", "dist"].map((name) => path.join(directory, "nested", name))
const files = roots.map((root) => path.join(root, "package", "index.js"))
yield* noUpdate(
(event) => roots.some((root) => event.file === root || event.file.startsWith(`${root}${path.sep}`)),
Effect.forEach(files, (file) => afs.writeWithDirs(file, "ignored"), {
concurrency: "unbounded",
discard: true,
}),
)
}),
withTmp(
(directory) =>
Effect.gen(function* () {
const afs = yield* FSUtil.Service
yield* ready(directory)
const roots = ["node_modules", ".git", "dist"].map((name) => path.join(directory, "nested", name))
const files = roots.map((root) => path.join(root, "package", "index.js"))
yield* noUpdate(
(event) => roots.some((root) => event.file === root || event.file.startsWith(`${root}${path.sep}`)),
Effect.forEach(files, (file) => afs.writeWithDirs(file, "ignored"), {
concurrency: "unbounded",
discard: true,
}),
)
}),
{ git: true },
),
)
@ -225,7 +223,10 @@ describeWatcher("LocationWatcher", () => {
Effect.promise(() => tmpdir()),
(tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
)
yield* ready(tmp.path).pipe(provide(tmp.path), Effect.scoped)
yield* ready(tmp.path).pipe(
provide(tmp.path, { type: "git", store: AbsolutePath.make(path.join(tmp.path, ".git")) }),
Effect.scoped,
)
const file = path.join(tmp.path, "after-dispose.txt")
yield* noUpdate((event) => event.file === file, fs.writeFileString(file, "gone")).pipe(
Effect.provideService(EventV2.Service, events),

View file

@ -14,6 +14,7 @@ const formID = Form.ID.create("frm_test")
const input = {
id: formID,
sessionID: SessionSchema.ID.make("ses_test"),
title: "Test form",
mode: "form",
fields: [{ key: "name", type: "string", required: true }],
} satisfies Form.CreateInput
@ -45,10 +46,12 @@ describe("Form", () => {
const service = yield* Form.Service
const created = yield* service.create({
sessionID: "global",
title: "MCP input",
mode: "form",
fields: [{ key: "name", type: "string", required: true }],
})
expect(created.sessionID).toBe("global")
expect(created.title).toBe("MCP input")
const owned = yield* service.list({ sessionID: "global" })
expect(owned.map((form) => form.id)).toEqual([created.id])
@ -64,6 +67,7 @@ describe("Form", () => {
const service = yield* Form.Service
const created = yield* service.create({
sessionID: "global",
title: "Conditional form",
mode: "form",
fields: [
{ key: "confirm", type: "boolean", required: true },
@ -96,6 +100,7 @@ describe("Form", () => {
]
const created = yield* service.create({
sessionID: "global",
title: "Multiselect form",
mode: "form",
fields: [
{ key: "langs", type: "multiselect", options },
@ -118,6 +123,7 @@ describe("Form", () => {
const service = yield* Form.Service
const created = yield* service.create({
sessionID: "global",
title: "Dependent form",
mode: "form",
fields: [
{ key: "a", type: "boolean" },
@ -160,6 +166,7 @@ describe("Form", () => {
]
const created = yield* service.create({
sessionID: "global",
title: "Selection form",
mode: "form",
fields: [
{ key: "langs", type: "multiselect", options },
@ -191,6 +198,7 @@ describe("Form", () => {
const service = yield* Form.Service
const created = yield* service.create({
sessionID: "global",
title: "Cascading form",
mode: "form",
fields: [
{ key: "a", type: "boolean" },
@ -213,7 +221,7 @@ describe("Form", () => {
Effect.gen(function* () {
const service = yield* Form.Service
const flipCreate = (fields: ReadonlyArray<Form.Field>) =>
service.create({ sessionID: "global", mode: "form", fields }).pipe(Effect.flip)
service.create({ sessionID: "global", title: "Invalid form", mode: "form", fields }).pipe(Effect.flip)
expect(
yield* flipCreate([

View file

@ -1,14 +1,33 @@
import { describe, expect } from "bun:test"
import { Duration, Effect, Exit, Fiber, Scope, Stream } from "effect"
import { Cause, Duration, Effect, Exit, Fiber, Layer, Scope, Stream } from "effect"
import * as TestClock from "effect/testing/TestClock"
import { Credential } from "@opencode-ai/core/credential"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
import { makeGlobalNode } from "@opencode-ai/core/effect/app-node"
import { LayerNode } from "@opencode-ai/core/effect/layer-node"
import { EventV2 } from "@opencode-ai/core/event"
import { Integration } from "@opencode-ai/core/integration"
import { testEffect } from "./lib/effect"
const it = testEffect(AppNodeBuilder.build(LayerNode.group([Integration.node, Credential.node, EventV2.node])))
const failingCredentialNode = makeGlobalNode({
service: Credential.Service,
layer: Layer.succeed(
Credential.Service,
Credential.Service.of({
all: () => Effect.succeed([]),
list: () => Effect.succeed([]),
get: () => Effect.succeed(undefined),
create: () => Effect.die(new Error("credential persistence failed")),
update: () => Effect.void,
remove: () => Effect.void,
}),
),
deps: [],
})
const failingIt = testEffect(
AppNodeBuilder.build(LayerNode.group([Integration.node, EventV2.node]), [[Credential.node, failingCredentialNode]]),
)
describe("Integration", () => {
it.effect("registers integrations through the editor", () =>
@ -254,6 +273,47 @@ describe("Integration", () => {
}),
)
failingIt.effect("fails the attempt when credential persistence fails", () =>
Effect.gen(function* () {
const integrations = yield* Integration.Service
const integrationID = Integration.ID.make("openai")
const methodID = Integration.MethodID.make("chatgpt")
yield* integrations.transform((editor) =>
editor.method.update({
integrationID,
method: { id: methodID, type: "oauth", label: "ChatGPT" },
authorize: () =>
Effect.succeed({
mode: "code" as const,
url: "https://example.com/authorize",
instructions: "Paste the code",
callback: () =>
Effect.succeed(
Credential.OAuth.make({
type: "oauth",
methodID,
access: "access",
refresh: "refresh",
expires: 1,
}),
),
}),
}),
)
const attempt = yield* integrations.connection.oauth({ integrationID, methodID, inputs: {} })
const exit = yield* integrations.attempt
.complete({ attemptID: attempt.attemptID, code: "1234" })
.pipe(Effect.exit)
expect(Exit.isFailure(exit) && Cause.hasDies(exit.cause)).toBe(true)
expect(yield* integrations.attempt.status(attempt.attemptID)).toEqual({
status: "failed",
message: "credential persistence failed",
time: attempt.time,
})
}),
)
it.effect("expires abandoned OAuth attempts", () =>
Effect.gen(function* () {
const integrations = yield* Integration.Service

View file

@ -92,6 +92,73 @@ describe("OpencodePlugin", () => {
}),
)
it.live("uses a canonical custom server throughout device authorization", () =>
Effect.acquireUseRelease(
Effect.sync(() => {
const requests: string[] = []
const server = Bun.serve({
port: 0,
fetch: (request) => {
const url = new URL(request.url)
requests.push(`${request.method} ${url.pathname}`)
if (url.pathname.endsWith("/auth/device/code")) {
return Response.json({
device_code: "device",
user_code: "user",
verification_uri_complete: `${url.origin}/verify`,
expires_in: 60,
interval: 0,
})
}
if (url.pathname.endsWith("/auth/device/token")) {
return Response.json({ access_token: "access", refresh_token: "refresh", expires_in: 600 })
}
if (url.pathname.endsWith("/api/user")) return Response.json({ id: "user", email: "user@example.com" })
if (url.pathname.endsWith("/api/orgs")) return Response.json([{ id: "org", name: "Org" }])
return new Response("Not found", { status: 404 })
},
})
return { requests, server }
}),
({ requests, server }) =>
Effect.gen(function* () {
yield* addPlugin()
const integrations = yield* Integration.Service
const attempt = yield* integrations.connection.oauth({
integrationID: Integration.ID.make("opencode"),
methodID: Integration.MethodID.make("device"),
inputs: { server: `${server.url.origin}/console///?ignored=true#ignored` },
})
expect(attempt.url).toBe(`${server.url.origin}/verify`)
yield* eventually(integrations.attempt.status(attempt.attemptID), (status) => status.status === "complete")
expect(requests).toContain("POST /console/auth/device/code")
expect(requests).toContain("POST /console/auth/device/token")
expect(requests).toContain("GET /console/api/user")
expect(requests).toContain("GET /console/api/orgs")
expect((yield* (yield* Credential.Service).list(Integration.ID.make("opencode")))[0]?.value).toMatchObject({
metadata: { server: `${server.url.origin}/console` },
})
}),
({ server }) => Effect.promise(() => server.stop(true)),
),
)
it.effect("rejects non-HTTP OpenCode servers", () =>
Effect.gen(function* () {
yield* addPlugin()
const error = yield* (yield* Integration.Service).connection
.oauth({
integrationID: Integration.ID.make("opencode"),
methodID: Integration.MethodID.make("device"),
inputs: { server: "ftp://console.example.com" },
})
.pipe(Effect.flip)
expect(error).toBeInstanceOf(Integration.AuthorizationError)
expect(String(error.cause)).toContain("Invalid OpenCode server URL: expected HTTP(S)")
}),
)
it.live("loads providers and models from the connected OpenCode server", () =>
Effect.acquireUseRelease(
Effect.sync(() => {

View file

@ -17,7 +17,6 @@ import { SessionCompaction } from "@opencode-ai/core/session/compaction"
import { SessionEvent } from "@opencode-ai/core/session/event"
import { SessionInput } from "@opencode-ai/core/session/input"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { Prompt } from "@opencode-ai/schema/prompt"
import { SessionProjector } from "@opencode-ai/core/session/projector"
import { SessionExecution } from "@opencode-ai/core/session/execution"
import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model"
@ -84,14 +83,16 @@ describe("SessionV2.compact", () => {
const created = yield* session.create({ location })
const messageID = SessionMessage.ID.create()
const prompt = Prompt.make({ text: "Please compact this session history." })
yield* events.publish(SessionEvent.PromptAdmitted, {
yield* events.publish(SessionEvent.InputAdmitted, {
sessionID: created.id,
inputID: messageID,
prompt,
delivery: "steer",
input: {
type: "user",
data: { text: "Please compact this session history." },
delivery: "steer",
},
})
yield* events.publish(SessionEvent.PromptPromoted, {
yield* events.publish(SessionEvent.InputPromoted, {
sessionID: created.id,
inputID: messageID,
})

View file

@ -17,7 +17,6 @@ import { ProviderV2 } from "@opencode-ai/core/provider"
import { AbsolutePath } from "@opencode-ai/core/schema"
import { SessionV2 } from "@opencode-ai/core/session"
import { SessionV1 } from "@opencode-ai/core/v1/session"
import { PromptInput } from "@opencode-ai/schema/prompt-input"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { SessionProjector } from "@opencode-ai/core/session/projector"
import { SessionExecution } from "@opencode-ai/core/session/execution"
@ -193,14 +192,12 @@ describe("SessionV2.create", () => {
const parent = yield* session.create({ location, title: "Parent" })
const admitted = yield* session.prompt({
sessionID: parent.id,
prompt: PromptInput.Prompt.make({ text: "First" }),
text: "First",
resume: false,
})
yield* SessionInput.promoteSteers(db, events, parent.id)
yield* events.publish(SessionEvent.Synthetic, {
sessionID: parent.id,
text: "parent note",
})
yield* session.synthetic({ sessionID: parent.id, text: "parent note", resume: false })
yield* SessionInput.promoteSteers(db, events, parent.id)
const forked = yield* session.fork({ sessionID: parent.id })
const parentContext = yield* session.context(parent.id)
@ -222,19 +219,25 @@ describe("SessionV2.create", () => {
})
expect(yield* SessionInput.find(db, forkContext[0].id)).toMatchObject({
sessionID: forked.id,
prompt: { text: "First" },
type: "user",
data: { text: "First" },
promotedSeq: 2,
})
expect(yield* SessionInput.find(db, forkContext[1].id)).toMatchObject({
sessionID: forked.id,
type: "synthetic",
data: { text: "parent note" },
})
yield* session.prompt({
sessionID: parent.id,
prompt: PromptInput.Prompt.make({ text: "Parent changed" }),
text: "Parent changed",
resume: false,
})
yield* SessionInput.promoteSteers(db, events, parent.id)
yield* session.prompt({
sessionID: forked.id,
prompt: PromptInput.Prompt.make({ text: "Child continues" }),
text: "Child continues",
resume: false,
})
yield* SessionInput.promoteSteers(db, events, forked.id)
@ -246,7 +249,7 @@ describe("SessionV2.create", () => {
Array.from(yield* Stream.runCollect(logEvents(session, forked.id))).map(
(event): number | undefined => event.durable?.seq,
),
).toEqual([0, 4, 5])
).toEqual([0, 5, 6])
expect(yield* SessionInput.find(db, admitted.id)).toMatchObject({ sessionID: parent.id })
}),
)
@ -259,13 +262,13 @@ describe("SessionV2.create", () => {
const parent = yield* session.create({ location })
const first = yield* session.prompt({
sessionID: parent.id,
prompt: PromptInput.Prompt.make({ text: "First" }),
text: "First",
resume: false,
})
yield* SessionInput.promoteSteers(db, events, parent.id)
const second = yield* session.prompt({
sessionID: parent.id,
prompt: PromptInput.Prompt.make({ text: "Second" }),
text: "Second",
resume: false,
})
yield* SessionInput.promoteSteers(db, events, parent.id)
@ -410,7 +413,7 @@ describe("SessionV2.create", () => {
const created = yield* session.create({ location })
yield* session.prompt({
sessionID: created.id,
prompt: PromptInput.Prompt.make({ text: "Hello" }),
text: "Hello",
resume: false,
})
yield* SessionInput.promoteSteers(db, events, created.id)
@ -418,8 +421,12 @@ describe("SessionV2.create", () => {
expect(
Array.from(yield* logEvents(session, created.id, true).pipe(Stream.take(2), Stream.runCollect)),
).toMatchObject([
{ durable: { seq: 1 }, type: "session.prompt.admitted", data: { prompt: { text: "Hello" } } },
{ durable: { seq: 2 }, type: "session.prompt.promoted" },
{
durable: { seq: 1 },
type: "session.input.admitted",
data: { input: { type: "user", data: { text: "Hello" }, delivery: "steer" } },
},
{ durable: { seq: 2 }, type: "session.input.promoted" },
])
}),
)
@ -432,7 +439,7 @@ describe("SessionV2.create", () => {
const created = yield* session.create({ id: SessionV2.ID.make("ses_fresh_target_replay"), location })
const admitted = yield* session.prompt({
sessionID: created.id,
prompt: PromptInput.Prompt.make({ text: "Replay lifecycle" }),
text: "Replay lifecycle",
resume: false,
})
yield* SessionInput.promoteSteers(sourceDb, sourceEvents, created.id)
@ -476,7 +483,8 @@ describe("SessionV2.create", () => {
expect(yield* SessionInput.find(db, admitted.id)).toMatchObject({
id: admitted.id,
sessionID: created.id,
prompt: { text: "Replay lifecycle" },
type: "user",
data: { text: "Replay lifecycle" },
delivery: "steer",
admittedSeq: 1,
})
@ -486,7 +494,8 @@ describe("SessionV2.create", () => {
expect(yield* SessionInput.find(db, admitted.id)).toMatchObject({
id: admitted.id,
sessionID: created.id,
prompt: { text: "Replay lifecycle" },
type: "user",
data: { text: "Replay lifecycle" },
delivery: "steer",
admittedSeq: 1,
promotedSeq: 2,
@ -504,8 +513,8 @@ describe("SessionV2.create", () => {
.pipe(Effect.orDie)).map((event) => [event.seq, event.type]),
).toEqual([
[0, EventV2.versionedType(SessionV1.Event.Created.type, 1)],
[1, EventV2.versionedType(SessionEvent.PromptAdmitted.type, 1)],
[2, EventV2.versionedType(SessionEvent.PromptPromoted.type, 1)],
[1, EventV2.versionedType(SessionEvent.InputAdmitted.type, 1)],
[2, EventV2.versionedType(SessionEvent.InputPromoted.type, 1)],
])
}).pipe(Effect.provide(Layer.fresh(targetLayer)))
}),

View file

@ -15,7 +15,6 @@ import { AbsolutePath } from "@opencode-ai/core/schema"
import { SessionV2 } from "@opencode-ai/core/session"
import { SessionEvent } from "@opencode-ai/core/session/event"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { Prompt } from "@opencode-ai/schema/prompt"
import { Money } from "@opencode-ai/schema/money"
import { SessionMessageUpdater } from "@opencode-ai/core/session/message-updater"
import { SessionProjector } from "@opencode-ai/core/session/projector"
@ -266,28 +265,26 @@ describe("SessionProjector", () => {
.pipe(Effect.orDie)
const events = yield* EventV2.Service
yield* events.publish(SessionEvent.PromptAdmitted, {
yield* events.publish(SessionEvent.InputAdmitted, {
sessionID,
inputID: SessionMessage.ID.make("msg_first"),
prompt: Prompt.make({ text: "first" }),
delivery: "steer",
input: { type: "user", data: { text: "first" }, delivery: "steer" },
})
yield* events.publish(
SessionEvent.PromptPromoted,
SessionEvent.InputPromoted,
{
sessionID,
inputID: SessionMessage.ID.make("msg_first"),
},
{ id: EventV2.ID.make("evt_z") },
)
yield* events.publish(SessionEvent.PromptAdmitted, {
yield* events.publish(SessionEvent.InputAdmitted, {
sessionID,
inputID: SessionMessage.ID.make("msg_second"),
prompt: Prompt.make({ text: "second" }),
delivery: "steer",
input: { type: "user", data: { text: "second" }, delivery: "steer" },
})
yield* events.publish(
SessionEvent.PromptPromoted,
SessionEvent.InputPromoted,
{
sessionID,
inputID: SessionMessage.ID.make("msg_second"),
@ -344,12 +341,11 @@ describe("SessionProjector", () => {
const admitted = yield* SessionInput.admit(db, events, {
id,
sessionID,
prompt: Prompt.make({ text: "promote me" }),
delivery: "steer",
input: { type: "user", data: { text: "promote me" }, delivery: "steer" },
})
if (!admitted) return yield* Effect.die("Prompt admission failed")
const event = yield* events.publish(SessionEvent.PromptPromoted, {
const event = yield* events.publish(SessionEvent.InputPromoted, {
sessionID,
inputID: id,
})

View file

@ -18,7 +18,6 @@ import { Project } from "@opencode-ai/core/project"
import { ProjectTable } from "@opencode-ai/core/project/sql"
import { AbsolutePath } from "@opencode-ai/core/schema"
import { SessionV2 } from "@opencode-ai/core/session"
import { PromptInput } from "@opencode-ai/schema/prompt-input"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { SessionProjector } from "@opencode-ai/core/session/projector"
import { SessionExecution } from "@opencode-ai/core/session/execution"
@ -174,16 +173,17 @@ describe("SessionV2.prompt", () => {
const message = yield* session.prompt({
sessionID,
prompt: PromptInput.Prompt.make({ text: "Fix the failing tests" }),
text: "Fix the failing tests",
resume: false,
})
expect(message.prompt.text).toBe("Fix the failing tests")
expect(message.data.text).toBe("Fix the failing tests")
expect(yield* session.messages({ sessionID })).toEqual([])
expect(yield* admitted(message.id)).toMatchObject({
id: message.id,
sessionID,
prompt: { text: "Fix the failing tests" },
type: "user",
data: { text: "Fix the failing tests" },
delivery: "steer",
})
}),
@ -198,7 +198,7 @@ describe("SessionV2.prompt", () => {
const boundary = yield* session.prompt({
sessionID,
prompt: PromptInput.Prompt.make({ text: "boundary" }),
text: "boundary",
resume: false,
})
yield* SessionInput.promoteSteers(db, events, sessionID)
@ -210,7 +210,7 @@ describe("SessionV2.prompt", () => {
})
expect((yield* session.get(sessionID)).revert?.messageID).toBe(boundary.id)
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "after revert" }), resume: false })
yield* session.prompt({ sessionID, text: "after revert", resume: false })
expect((yield* session.get(sessionID)).revert).toBeUndefined()
expect(
@ -222,6 +222,35 @@ describe("SessionV2.prompt", () => {
}),
)
it.effect("holds synthetic input behind a staged revert and discards it when committed", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
const events = yield* EventV2.Service
const { db } = yield* Database.Service
const boundary = yield* session.prompt({
sessionID,
text: "boundary",
resume: false,
})
yield* SessionInput.promoteSteers(db, events, sessionID)
yield* events.publish(SessionEvent.RevertEvent.Staged, {
sessionID,
revert: { messageID: boundary.id, files: [] },
})
wakeCalls.length = 0
const completion = yield* session.synthetic({ sessionID, text: "stale completion" })
expect(wakeCalls).toEqual([])
expect(yield* SessionInput.find(db, completion.id)).toMatchObject({ type: "synthetic" })
yield* session.revert.commit(sessionID)
expect(yield* SessionInput.find(db, completion.id)).toBeUndefined()
}),
)
it.effect("resolves attachment MIME before admission", () =>
Effect.gen(function* () {
yield* setup
@ -231,14 +260,12 @@ describe("SessionV2.prompt", () => {
const message = yield* session.prompt({
sessionID,
prompt: {
text: "Inspect this image",
files: [{ uri, name: "image.png", mention: { start: 8, end: 17, text: "[Image 1]" } }],
},
text: "Inspect this image",
files: [{ uri, name: "image.png", mention: { start: 8, end: 17, text: "[Image 1]" } }],
resume: false,
})
expect(message.prompt.files).toEqual([
expect(message.data.files).toEqual([
{
data: uri.slice(uri.indexOf(",") + 1),
mime: "image/png",
@ -248,8 +275,8 @@ describe("SessionV2.prompt", () => {
},
])
const stored = yield* admitted(message.id)
expect(stored?.type).toBe("prompt")
if (stored?.type === "prompt") expect(stored.prompt.files).toEqual(message.prompt.files)
expect(stored?.type).toBe("user")
if (stored?.type === "user") expect(stored.data.files).toEqual(message.data.files)
}),
)
@ -265,21 +292,19 @@ describe("SessionV2.prompt", () => {
const message = yield* session.prompt({
sessionID,
prompt: {
text: "Inspect this",
files: [{ uri: sourceUri.href, name: "main.ts" }],
},
text: "Inspect this",
files: [{ uri: sourceUri.href, name: "main.ts" }],
resume: false,
})
expect(message.prompt.files).toHaveLength(1)
expect(message.prompt.files?.[0]).toMatchObject({
expect(message.data.files).toHaveLength(1)
expect(message.data.files?.[0]).toMatchObject({
mime: "text/plain",
source: { type: "uri", uri: sourceUri.href },
name: "main.ts",
})
expect(
Buffer.from(message.prompt.files?.[0]?.data ?? "", "base64")
Buffer.from(message.data.files?.[0]?.data ?? "", "base64")
.toString("utf8")
.replace(/\r$/, ""),
).toBe('import { describe, expect } from "bun:test"')
@ -294,17 +319,18 @@ describe("SessionV2.prompt", () => {
const message = yield* session.prompt({
sessionID,
prompt: { text: "Inspect this", files: [{ uri, name: "source" }] },
text: "Inspect this",
files: [{ uri, name: "source" }],
resume: false,
})
expect(message.prompt.files).toHaveLength(1)
expect(message.prompt.files?.[0]).toMatchObject({
expect(message.data.files).toHaveLength(1)
expect(message.data.files?.[0]).toMatchObject({
mime: "application/x-directory",
source: { type: "uri", uri },
name: "source",
})
expect(Buffer.from(message.prompt.files?.[0]?.data ?? "", "base64").toString("utf8")).toContain(
expect(Buffer.from(message.data.files?.[0]?.data ?? "", "base64").toString("utf8")).toContain(
"session-prompt.test.ts",
)
}),
@ -327,11 +353,12 @@ describe("SessionV2.prompt", () => {
const message = yield* session.prompt({
sessionID,
prompt: { text: "Inspect this image", files: [{ uri: pathToFileURL(source).href }] },
text: "Inspect this image",
files: [{ uri: pathToFileURL(source).href }],
resume: false,
})
expect(message.prompt.files).toEqual([
expect(message.data.files).toEqual([
{
data: bytes.toString("base64"),
mime: "image/png",
@ -340,7 +367,7 @@ describe("SessionV2.prompt", () => {
},
])
const stored = yield* admitted(message.id)
expect(stored?.type === "prompt" ? stored.prompt.files : undefined).toEqual(message.prompt.files)
expect(stored?.type === "user" ? stored.data.files : undefined).toEqual(message.data.files)
}),
)
@ -352,11 +379,12 @@ describe("SessionV2.prompt", () => {
const message = yield* session.prompt({
sessionID,
prompt: { text: "Inspect this", files: [{ uri, name: "main.ts" }] },
text: "Inspect this",
files: [{ uri, name: "main.ts" }],
resume: false,
})
expect(message.prompt.files).toEqual([
expect(message.data.files).toEqual([
{
data: Buffer.from("export const value = 1\n").toString("base64"),
mime: "text/plain",
@ -376,7 +404,8 @@ describe("SessionV2.prompt", () => {
const error = yield* session
.prompt({
sessionID,
prompt: { text: "Inspect this", files: [{ uri, name: "image.png" }] },
text: "Inspect this",
files: [{ uri, name: "image.png" }],
resume: false,
})
.pipe(Effect.flip)
@ -402,22 +431,22 @@ describe("SessionV2.prompt", () => {
const fiber = yield* publicEvents({ sessionID }).pipe(Stream.take(4), Stream.runCollect, Effect.forkScoped)
yield* Effect.yieldNow
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "First" }), resume: false })
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Second" }), resume: false })
yield* session.prompt({ sessionID, text: "First", resume: false })
yield* session.prompt({ sessionID, text: "Second", resume: false })
yield* SessionInput.promoteSteers(db, events, sessionID)
const streamed = Array.from(yield* Fiber.join(fiber))
expect(streamed.map((event): [number | undefined, string] => [event.durable?.seq, event.type])).toEqual([
[0, "session.prompt.admitted"],
[1, "session.prompt.admitted"],
[2, "session.prompt.promoted"],
[3, "session.prompt.promoted"],
[0, "session.input.admitted"],
[1, "session.input.admitted"],
[2, "session.input.promoted"],
[3, "session.input.promoted"],
])
expect(
Array.from(
yield* publicEvents({ sessionID, after: streamed[0].durable?.seq }).pipe(Stream.take(1), Stream.runCollect),
).map((event): [number | undefined, string] => [event.durable?.seq, event.type]),
).toEqual([[1, "session.prompt.admitted"]])
).toEqual([[1, "session.input.admitted"]])
}),
)
@ -427,7 +456,7 @@ describe("SessionV2.prompt", () => {
const session = yield* SessionV2.Service
const message = yield* session.prompt({
sessionID,
prompt: PromptInput.Prompt.make({ text: "Fix the failing tests" }),
text: "Fix the failing tests",
resume: false,
})
@ -446,7 +475,7 @@ describe("SessionV2.prompt", () => {
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
const input = { sessionID, prompt: PromptInput.Prompt.make({ text: "Fix the failing tests" }), resume: false }
const input = { sessionID, text: "Fix the failing tests", resume: false }
const first = yield* session.prompt(input)
const second = yield* session.prompt(input)
@ -464,7 +493,7 @@ describe("SessionV2.prompt", () => {
const input = {
sessionID,
id: messageID,
prompt: PromptInput.Prompt.make({ text: "Fix the failing tests" }),
text: "Fix the failing tests",
resume: false,
}
@ -484,7 +513,7 @@ describe("SessionV2.prompt", () => {
const input = {
sessionID,
id: messageID,
prompt: PromptInput.Prompt.make({ text: "Recover committed prompt" }),
text: "Recover committed prompt",
resume: false,
}
const first = yield* session.prompt(input)
@ -505,13 +534,13 @@ describe("SessionV2.prompt", () => {
yield* session.prompt({
sessionID,
id: messageID,
prompt: PromptInput.Prompt.make({ text: "Fix the failing tests" }),
text: "Fix the failing tests",
})
const failure = yield* session
.prompt({
sessionID,
id: messageID,
prompt: PromptInput.Prompt.make({ text: "Delete the failing tests" }),
text: "Delete the failing tests",
resume: false,
})
.pipe(Effect.flip)
@ -530,14 +559,14 @@ describe("SessionV2.prompt", () => {
yield* session.prompt({
id: messageID,
sessionID,
prompt: PromptInput.Prompt.make({ text: "Fix the failing tests" }),
text: "Fix the failing tests",
resume: false,
})
const failure = yield* session
.prompt({
id: messageID,
sessionID,
prompt: PromptInput.Prompt.make({ text: "Fix the failing tests" }),
text: "Fix the failing tests",
delivery: "queue",
resume: false,
})
@ -554,7 +583,7 @@ describe("SessionV2.prompt", () => {
const input = {
sessionID,
id: messageID,
prompt: PromptInput.Prompt.make({ text: "Fix the failing tests" }),
text: "Fix the failing tests",
resume: false,
}
@ -563,7 +592,7 @@ describe("SessionV2.prompt", () => {
expect(messages[1]).toEqual(messages[0])
expect(yield* session.messages({ sessionID })).toEqual([])
expect(yield* admittedCount).toBe(1)
expect(yield* eventCount(EventV2.versionedType(SessionEvent.PromptAdmitted.type, 1))).toBe(1)
expect(yield* eventCount(EventV2.versionedType(SessionEvent.InputAdmitted.type, 1))).toBe(1)
}),
)
@ -576,7 +605,7 @@ describe("SessionV2.prompt", () => {
yield* session.prompt({
id: messageID,
sessionID,
prompt: PromptInput.Prompt.make({ text: "Promote once" }),
text: "Promote once",
resume: false,
})
@ -585,7 +614,7 @@ describe("SessionV2.prompt", () => {
{ concurrency: "unbounded" },
)
expect(yield* eventCount(EventV2.versionedType(SessionEvent.PromptPromoted.type, 1))).toBe(1)
expect(yield* eventCount(EventV2.versionedType(SessionEvent.InputPromoted.type, 1))).toBe(1)
expect(yield* admitted(messageID)).toMatchObject({ promotedSeq: 1 })
expect(yield* session.messages({ sessionID })).toMatchObject([
{ id: messageID, type: "user", text: "Promote once" },
@ -603,9 +632,11 @@ describe("SessionV2.prompt", () => {
yield* session.prompt({
id: messageID,
sessionID,
prompt: PromptInput.Prompt.make({ text: "Replay pending" }),
text: "Replay pending",
resume: false,
})
const syntheticID = SessionMessage.ID.create()
yield* session.synthetic({ id: syntheticID, sessionID, text: "Replay synthetic", resume: false })
const recorded = yield* db
.select()
.from(EventTable)
@ -631,7 +662,16 @@ describe("SessionV2.prompt", () => {
})),
)
expect(yield* admitted(messageID)).toMatchObject({ id: messageID, prompt: { text: "Replay pending" } })
expect(yield* admitted(messageID)).toMatchObject({
id: messageID,
type: "user",
data: { text: "Replay pending" },
})
expect(yield* admitted(syntheticID)).toMatchObject({
id: syntheticID,
type: "synthetic",
data: { text: "Replay synthetic" },
})
expect(yield* session.messages({ sessionID })).toEqual([])
expect(wakeCalls).toEqual([])
}),
@ -656,11 +696,9 @@ describe("SessionV2.prompt", () => {
.onConflictDoNothing()
.run()
.pipe(Effect.orDie)
const prompt = PromptInput.Prompt.make({ text: "Fix the failing tests" })
yield* session.prompt({ id: messageID, sessionID, prompt, resume: false })
yield* session.prompt({ id: messageID, sessionID, text: "Fix the failing tests", resume: false })
const failure = yield* session
.prompt({ id: messageID, sessionID: other, prompt, resume: false })
.prompt({ id: messageID, sessionID: other, text: "Fix the failing tests", resume: false })
.pipe(Effect.flip)
expect(failure).toMatchObject({ _tag: "Session.PromptConflictError", sessionID: other, messageID })
@ -692,7 +730,7 @@ describe("SessionV2.prompt", () => {
.prompt({
id: messageID,
sessionID,
prompt: PromptInput.Prompt.make({ text: "Conflicting prompt" }),
text: "Conflicting prompt",
resume: false,
})
.pipe(Effect.flip)
@ -709,7 +747,7 @@ describe("SessionV2.prompt", () => {
executionCalls.length = 0
wakeCalls.length = 0
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Run by default" }) })
yield* session.prompt({ sessionID, text: "Run by default" })
expect(executionCalls).toEqual([])
expect(wakeCalls).toEqual([sessionID])
@ -725,7 +763,7 @@ describe("SessionV2.prompt", () => {
yield* session.prompt({
sessionID,
prompt: PromptInput.Prompt.make({ text: "Run explicitly" }),
text: "Run explicitly",
resume: true,
})
@ -741,10 +779,150 @@ describe("SessionV2.prompt", () => {
executionCalls.length = 0
wakeCalls.length = 0
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Do not run" }), resume: false })
yield* session.prompt({ sessionID, text: "Do not run", resume: false })
expect(executionCalls).toEqual([])
expect(wakeCalls).toEqual([])
}),
)
it.effect("treats prompt metadata as durable retry identity", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
const input = {
id: messageID,
sessionID,
text: "Deploy",
metadata: { source: "api" },
resume: false,
}
const first = yield* session.prompt(input)
const retried = yield* session.prompt(input)
const failure = yield* session.prompt({ ...input, metadata: { source: "plugin" } }).pipe(Effect.flip)
expect(retried).toEqual(first)
expect(first.data.metadata).toEqual({ source: "api" })
expect(failure._tag).toBe("Session.PromptConflictError")
}),
)
it.effect("durably admits synthetic input before transcript promotion", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
const events = yield* EventV2.Service
const { db } = yield* Database.Service
const input = yield* session.synthetic({
id: messageID,
sessionID,
text: "Background work completed",
description: "shell completion",
metadata: { job: "shell" },
resume: false,
})
expect(yield* session.messages({ sessionID })).toEqual([])
expect(yield* admitted(input.id)).toMatchObject({
type: "synthetic",
sessionID,
delivery: "steer",
data: {
text: "Background work completed",
description: "shell completion",
metadata: { job: "shell" },
},
})
yield* SessionInput.promoteSteers(db, events, sessionID)
expect(yield* session.messages({ sessionID })).toMatchObject([
{
id: messageID,
type: "synthetic",
text: "Background work completed",
description: "shell completion",
metadata: { job: "shell" },
},
])
}),
)
it.effect("reconciles exact synthetic retries and rejects conflicting reuse", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
const events = yield* EventV2.Service
const database = yield* Database.Service
const input = { id: messageID, sessionID, text: "Completed", resume: false }
const entries = yield* Effect.all([session.synthetic(input), session.synthetic(input)], {
concurrency: "unbounded",
})
yield* SessionInput.promoteSteers(database.db, events, sessionID)
const promotedRetry = yield* session.synthetic(input)
const failure = yield* session.synthetic({ ...input, text: "Different completion" }).pipe(Effect.flip)
expect(entries[1]).toEqual(entries[0])
expect(promotedRetry).toMatchObject({ id: messageID, type: "synthetic", promotedSeq: expect.any(Number) })
expect(failure).toMatchObject({ _tag: "Session.SyntheticConflictError", sessionID, inputID: messageID })
expect(yield* admittedCount).toBe(1)
expect(yield* eventCount(EventV2.versionedType(SessionEvent.InputAdmitted.type, 1))).toBe(1)
}),
)
it.effect("keeps synthetic queue input pending until the queue boundary", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
const events = yield* EventV2.Service
const { db } = yield* Database.Service
const input = yield* session.synthetic({
sessionID,
text: "Queued completion",
delivery: "queue",
resume: false,
})
expect(input.delivery).toBe("queue")
expect(yield* SessionInput.promoteSteers(db, events, sessionID)).toBe(0)
expect(yield* session.messages({ sessionID })).toEqual([])
expect(yield* SessionInput.promoteNextQueued(db, events, sessionID)).toBe(true)
expect(yield* session.messages({ sessionID })).toMatchObject([
{ id: input.id, type: "synthetic", text: "Queued completion" },
])
}),
)
it.effect("promotes prompt and synthetic steers in admission order", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
const events = yield* EventV2.Service
const { db } = yield* Database.Service
yield* session.prompt({
sessionID,
text: "First prompt",
resume: false,
})
yield* session.synthetic({ sessionID, text: "Background completion", resume: false })
yield* session.prompt({
sessionID,
text: "Second prompt",
resume: false,
})
yield* SessionInput.promoteSteers(db, events, sessionID)
expect(
(yield* session.messages({ sessionID, order: "asc" })).map((message) =>
message.type === "user" || message.type === "synthetic" ? message.text : message.type,
),
).toEqual(["First prompt", "Background completion", "Second prompt"])
}),
)
})

View file

@ -24,7 +24,6 @@ import { SessionV2 } from "@opencode-ai/core/session"
import { Snapshot } from "@opencode-ai/core/snapshot"
import { SessionCompaction } from "@opencode-ai/core/session/compaction"
import { SessionTitle } from "@opencode-ai/core/session/title"
import { PromptInput } from "@opencode-ai/schema/prompt-input"
import { SessionProjector } from "@opencode-ai/core/session/projector"
import { SessionExecution } from "@opencode-ai/core/session/execution"
import { SessionRunCoordinator } from "@opencode-ai/core/session/run-coordinator"
@ -205,7 +204,7 @@ describe("SessionRunnerLLM recorded", () => {
const session = yield* SessionV2.Service
const prompt = yield* session.prompt({
sessionID,
prompt: PromptInput.Prompt.make({ text: "Say hello in one short sentence." }),
text: "Say hello in one short sentence.",
resume: false,
})
@ -236,8 +235,8 @@ describe("SessionRunnerLLM recorded", () => {
.orderBy(EventTable.seq)
.all()).map((event) => event.type),
).toEqual([
"session.prompt.admitted.1",
"session.prompt.promoted.1",
"session.input.admitted.1",
"session.input.promoted.1",
"session.step.started.1",
"session.text.started.1",
"session.text.ended.1",

View file

@ -65,7 +65,6 @@ import { Snapshot } from "@opencode-ai/core/snapshot"
import { SessionEvent } from "@opencode-ai/core/session/event"
import { SessionInput } from "@opencode-ai/core/session/input"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { PromptInput } from "@opencode-ai/schema/prompt-input"
import { Money } from "@opencode-ai/schema/money"
import { SessionProjector } from "@opencode-ai/core/session/projector"
import { SessionExecution } from "@opencode-ai/core/session/execution"
@ -109,6 +108,7 @@ const requests: LLMRequest[] = []
let response: LLMEvent[] = []
let responses: LLMEvent[][] | undefined
let responseStream: Stream.Stream<LLMEvent, LLMError> | undefined
let responseStreams: Stream.Stream<LLMEvent, LLMError>[] | undefined
let streamGate: Deferred.Deferred<void> | undefined
let streamStarted: Deferred.Deferred<void> | undefined
let streamFailure: LLMError | undefined
@ -123,6 +123,7 @@ const client = Layer.succeed(
prepare: () => Effect.die("unused"),
stream: ((request: LLMRequest) => {
requests.push(request)
if (responseStreams) return responseStreams.shift() ?? Stream.empty
if (responseStream) {
const stream = responseStream
responseStream = undefined
@ -444,8 +445,7 @@ const it = testEffect(
)
const sessionID = SessionV2.ID.make("ses_runner_test")
const otherSessionID = SessionV2.ID.make("ses_runner_other")
const admit = (session: SessionV2.Interface, text: string) =>
session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text }), resume: false })
const admit = (session: SessionV2.Interface, text: string) => session.prompt({ sessionID, text, resume: false })
const insertSession = (id: SessionV2.ID) =>
Effect.gen(function* () {
@ -482,6 +482,7 @@ const setup = Effect.gen(function* () {
responses = undefined
streamFailure = undefined
responseStream = undefined
responseStreams = undefined
streamGate = undefined
streamStarted = undefined
toolExecutionGate = undefined
@ -781,7 +782,7 @@ describe("SessionRunnerLLM", () => {
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Use echo" }), resume: false })
yield* session.prompt({ sessionID, text: "Use echo", resume: false })
const usage = new Usage({
inputTokens: 8,
outputTokens: 3,
@ -856,7 +857,7 @@ describe("SessionRunnerLLM", () => {
})
yield* session.prompt({
sessionID: child.id,
prompt: PromptInput.Prompt.make({ text: "Explore" }),
text: "Explore",
resume: false,
})
@ -898,7 +899,7 @@ describe("SessionRunnerLLM", () => {
const session = yield* SessionV2.Service
yield* session.prompt({
sessionID,
prompt: PromptInput.Prompt.make({ text: "Use echo twice" }),
text: "Use echo twice",
resume: false,
})
responses = [
@ -933,7 +934,7 @@ describe("SessionRunnerLLM", () => {
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Fail tool" }), resume: false })
yield* session.prompt({ sessionID, text: "Fail tool", resume: false })
responses = [
[
LLMEvent.stepStart({ index: 0 }),
@ -1010,7 +1011,7 @@ describe("SessionRunnerLLM", () => {
const message = yield* session.prompt({
sessionID,
prompt: PromptInput.Prompt.make({ text: "Run automatically" }),
text: "Run automatically",
})
yield* session.wait(sessionID)
@ -1021,6 +1022,34 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("runs a follow-up when synthetic input arrives during an active continuation", () =>
Effect.gen(function* () {
const session = yield* setup
const secondStarted = yield* Deferred.make<void>()
const releaseSecond = yield* Deferred.make<void>()
responseStreams = [
Stream.fromIterable(reply.tool("call-echo", "echo", { text: "background started" })),
Stream.unwrap(
Deferred.succeed(secondStarted, undefined).pipe(
Effect.andThen(Deferred.await(releaseSecond)),
Effect.as(Stream.fromIterable(reply.stop())),
),
),
Stream.fromIterable(reply.text("Handled completion", "text-completion")),
]
yield* admit(session, "Start background work")
const running = yield* session.resume(sessionID).pipe(Effect.forkChild({ startImmediately: true }))
yield* Deferred.await(secondStarted)
yield* session.synthetic({ sessionID, text: "Background work completed" })
yield* Deferred.succeed(releaseSecond, undefined)
yield* Fiber.join(running)
expect(requests).toHaveLength(3)
expect(userTexts(requests[2]!)).toContain("Background work completed")
}),
)
it.effect("streams one request with registry definitions from chronological V2 user history", () =>
Effect.gen(function* () {
const session = yield* setup
@ -1049,7 +1078,7 @@ describe("SessionRunnerLLM", () => {
yield* session.prompt({
id: messageID,
sessionID,
prompt: PromptInput.Prompt.make({ text: "First" }),
text: "First",
resume: false,
})
@ -1068,7 +1097,7 @@ describe("SessionRunnerLLM", () => {
).toBeUndefined()
systemUnavailable = false
yield* session.prompt({ id: messageID, sessionID, prompt: PromptInput.Prompt.make({ text: "First" }) })
yield* session.prompt({ id: messageID, sessionID, text: "First" })
yield* session.wait(sessionID)
expect(requests).toHaveLength(1)
@ -1335,7 +1364,7 @@ describe("SessionRunnerLLM", () => {
.run()
.pipe(Effect.orDie)
const session = yield* SessionV2.Service
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Inspect files" }), resume: false })
yield* session.prompt({ sessionID, text: "Inspect files", resume: false })
requests.length = 0
response = []
@ -1356,7 +1385,7 @@ describe("SessionRunnerLLM", () => {
const release = yield* Deferred.make<void>()
pluginFlushHook = Deferred.await(release)
const session = yield* SessionV2.Service
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Wait for plugins" }), resume: false })
yield* session.prompt({ sessionID, text: "Wait for plugins", resume: false })
requests.length = 0
response = []
@ -1645,9 +1674,10 @@ describe("SessionRunnerLLM", () => {
expect((yield* session.messages({ sessionID })).find((message) => message.id === first.id)).toBeUndefined()
yield* admit(session, "Steer after compaction")
yield* session.synthetic({ sessionID, text: "Completion after compaction", resume: false })
yield* session.prompt({
sessionID,
prompt: PromptInput.Prompt.make({ text: "Queue after compaction" }),
text: "Queue after compaction",
delivery: "queue",
resume: false,
})
@ -1659,6 +1689,7 @@ describe("SessionRunnerLLM", () => {
expect(requests).toHaveLength(4)
expect(userTexts(requests[1])[0]).toContain("Create a new anchored summary")
expect(userTexts(requests[2])).toContain("Steer after compaction")
expect(userTexts(requests[2])).toContain("Completion after compaction")
expect(userTexts(requests[3])).toContain("Queue after compaction")
expect(yield* SessionInput.pendingCompaction((yield* Database.Service).db, sessionID)).toBeUndefined()
expect((yield* session.messages({ sessionID })).find((message) => message.id === first.id)).toMatchObject({
@ -1691,7 +1722,7 @@ describe("SessionRunnerLLM", () => {
const compaction = yield* session.compact({ sessionID })
yield* session.prompt({
sessionID,
prompt: PromptInput.Prompt.make({ text: "Continue after failure" }),
text: "Continue after failure",
delivery: "queue",
resume: false,
})
@ -2447,7 +2478,7 @@ describe("SessionRunnerLLM", () => {
const first = yield* session.resume(sessionID).pipe(Effect.forkChild)
yield* Deferred.await(streamStarted)
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Change direction" }) })
yield* session.prompt({ sessionID, text: "Change direction" })
yield* Deferred.succeed(streamGate, undefined)
yield* Fiber.join(first)
streamGate = undefined
@ -2480,7 +2511,7 @@ describe("SessionRunnerLLM", () => {
yield* Deferred.await(streamStarted)
yield* session.prompt({
sessionID,
prompt: PromptInput.Prompt.make({ text: "Wait until continuation ends" }),
text: "Wait until continuation ends",
delivery: "queue",
})
yield* Deferred.succeed(streamGate, undefined)
@ -2512,7 +2543,7 @@ describe("SessionRunnerLLM", () => {
yield* Deferred.await(streamStarted)
yield* session.prompt({
sessionID,
prompt: PromptInput.Prompt.make({ text: "Run after interrupt" }),
text: "Run after interrupt",
delivery: "queue",
})
yield* session.interrupt(sessionID)
@ -2546,7 +2577,7 @@ describe("SessionRunnerLLM", () => {
yield* Deferred.await(streamStarted)
yield* session.prompt({
sessionID,
prompt: PromptInput.Prompt.make({ text: "Steer after interrupt" }),
text: "Steer after interrupt",
})
yield* session.interrupt(sessionID)
expect(yield* Fiber.await(run)).toMatchObject({ _tag: "Failure" })
@ -2577,8 +2608,8 @@ describe("SessionRunnerLLM", () => {
const first = yield* session.resume(sessionID).pipe(Effect.forkChild)
yield* Deferred.await(streamStarted)
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Queue first" }), delivery: "queue" })
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Queue second" }), delivery: "queue" })
yield* session.prompt({ sessionID, text: "Queue first", delivery: "queue" })
yield* session.prompt({ sessionID, text: "Queue second", delivery: "queue" })
yield* Deferred.succeed(streamGate, undefined)
yield* Fiber.join(first)
streamGate = undefined
@ -2602,7 +2633,7 @@ describe("SessionRunnerLLM", () => {
yield* admit(session, "Start steering")
yield* session.prompt({
sessionID,
prompt: PromptInput.Prompt.make({ text: "Queue for later" }),
text: "Queue for later",
delivery: "queue",
resume: false,
})
@ -2629,16 +2660,17 @@ describe("SessionRunnerLLM", () => {
const first = yield* session.resume(sessionID).pipe(Effect.forkChild)
while (requests.length < 1) yield* Effect.yieldNow
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Queue first" }), delivery: "queue" })
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Queue second" }), delivery: "queue" })
yield* session.prompt({ sessionID, text: "Queue first", delivery: "queue" })
yield* session.prompt({ sessionID, text: "Queue second", delivery: "queue" })
streamGate = secondGate
yield* Deferred.succeed(firstGate, undefined)
while (requests.length < 2) yield* Effect.yieldNow
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Steer before next queued input" }) })
yield* session.prompt({ sessionID, text: "Steer before next queued input" })
yield* session.prompt({
sessionID,
prompt: PromptInput.Prompt.make({ text: "Also steer before next queued input" }),
text: "Also steer before next queued input",
})
yield* session.synthetic({ sessionID, text: "Background completion before next queued input" })
yield* Deferred.succeed(secondGate, undefined)
yield* Fiber.join(first)
streamGate = undefined
@ -2651,12 +2683,14 @@ describe("SessionRunnerLLM", () => {
"Queue first",
"Steer before next queued input",
"Also steer before next queued input",
"Background completion before next queued input",
])
expect(userTexts(requests[3]!)).toEqual([
"Start working",
"Queue first",
"Steer before next queued input",
"Also steer before next queued input",
"Background completion before next queued input",
"Queue second",
])
}),
@ -2673,8 +2707,8 @@ describe("SessionRunnerLLM", () => {
const first = yield* session.resume(sessionID).pipe(Effect.forkChild)
yield* Deferred.await(streamStarted)
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "First steer" }) })
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Second steer" }) })
yield* session.prompt({ sessionID, text: "First steer" })
yield* session.prompt({ sessionID, text: "Second steer" })
yield* Deferred.succeed(streamGate, undefined)
yield* Fiber.join(first)
streamGate = undefined
@ -2700,7 +2734,7 @@ describe("SessionRunnerLLM", () => {
const first = yield* session.resume(sessionID).pipe(Effect.forkChild)
yield* Deferred.await(streamStarted)
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Recover with this" }) })
yield* session.prompt({ sessionID, text: "Recover with this" })
yield* Deferred.succeed(streamGate, undefined)
expect(yield* Fiber.join(first).pipe(Effect.flip)).toBe(streamFailure)
@ -2859,7 +2893,7 @@ describe("SessionRunnerLLM", () => {
const session = yield* setup
yield* session.prompt({
sessionID,
prompt: PromptInput.Prompt.make({ text: "Wait in queue" }),
text: "Wait in queue",
delivery: "queue",
resume: false,
})
@ -2878,7 +2912,7 @@ describe("SessionRunnerLLM", () => {
const events = yield* EventV2.Service
const defect = new Error("fail after prompt promotion")
let fail = true
yield* events.project(SessionEvent.PromptPromoted, () => (fail ? Effect.die(defect) : Effect.void))
yield* events.project(SessionEvent.InputPromoted, () => (fail ? Effect.die(defect) : Effect.void))
yield* admit(session, "Recover promoted input")
expect(yield* session.resume(sessionID).pipe(Effect.catchDefect(Effect.succeed))).toBe(defect)
@ -2898,7 +2932,7 @@ describe("SessionRunnerLLM", () => {
const session = yield* setup
const events = yield* EventV2.Service
yield* events.listen((event) =>
event.type === SessionEvent.PromptPromoted.type
event.type === SessionEvent.InputPromoted.type
? Effect.die("fail after prompt promotion commits")
: Effect.void,
)
@ -2918,7 +2952,7 @@ describe("SessionRunnerLLM", () => {
yield* admit(session, "Run first")
yield* session.prompt({
sessionID: otherSessionID,
prompt: PromptInput.Prompt.make({ text: "Run second" }),
text: "Run second",
resume: false,
})
@ -2953,12 +2987,12 @@ describe("SessionRunnerLLM", () => {
yield* insertSession(otherLongSessionID)
yield* session.prompt({
sessionID: longSessionID,
prompt: PromptInput.Prompt.make({ text: "Run long session" }),
text: "Run long session",
resume: false,
})
yield* session.prompt({
sessionID: otherLongSessionID,
prompt: PromptInput.Prompt.make({ text: "Run other long session" }),
text: "Run other long session",
resume: false,
})
@ -3523,7 +3557,7 @@ describe("SessionRunnerLLM", () => {
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
yield* Deferred.await(streamStarted)
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Change direction" }) })
yield* session.prompt({ sessionID, text: "Change direction" })
yield* Deferred.succeed(streamGate, undefined)
yield* Fiber.join(run)
streamGate = undefined

View file

@ -9,7 +9,6 @@ import { LayerNode } from "@opencode-ai/core/effect/layer-node"
import { EventV2 } from "@opencode-ai/core/event"
import { SessionEvent } from "@opencode-ai/core/session/event"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { Prompt } from "@opencode-ai/schema/prompt"
import { SessionProjector } from "@opencode-ai/core/session/projector"
import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model"
import { SessionTable } from "@opencode-ai/core/session/sql"
@ -84,13 +83,12 @@ const prompt = (sessionID: SessionV2.ID, text: string) =>
Effect.gen(function* () {
const events = yield* EventV2.Service
const messageID = SessionMessage.ID.create()
yield* events.publish(SessionEvent.PromptAdmitted, {
yield* events.publish(SessionEvent.InputAdmitted, {
sessionID,
inputID: messageID,
prompt: Prompt.make({ text }),
delivery: "steer",
input: { type: "user", data: { text }, delivery: "steer" },
})
yield* events.publish(SessionEvent.PromptPromoted, {
yield* events.publish(SessionEvent.InputPromoted, {
sessionID,
inputID: messageID,
})

View file

@ -137,7 +137,9 @@ test("Core reuses the canonical shared schemas", async () => {
[SessionV2.Info, Session.Info],
[SessionV2.ListAnchor, Session.ListAnchor],
[coreSessionInput.Delivery, SessionInput.Delivery],
[coreSessionInput.Admitted, SessionInput.Admitted],
[coreSessionInput.Message, SessionInput.Message],
[coreSessionInput.User, SessionInput.User],
[coreSessionInput.Synthetic, SessionInput.Synthetic],
[coreSessionMessage.ID, SessionMessage.ID],
[coreSessionMessage.AssistantRetry, SessionMessage.AssistantRetry],
[coreSessionMessage.AgentSelected, SessionMessage.AgentSelected],

View file

@ -156,6 +156,7 @@ describe("QuestionTool", () => {
expect(assertions).toMatchObject([{ sessionID, action: "question", resources: ["*"] }])
expect(capturedInput()).toEqual({
sessionID,
title: "Questions",
metadata: { kind: "question", tool: { messageID: toolIdentity.assistantMessageID, callID: "call-question" } },
mode: "form",
fields: [
@ -202,6 +203,7 @@ describe("QuestionTool", () => {
})
expect(capturedInput()).toEqual({
sessionID,
title: "Questions",
metadata: { kind: "question", tool: { messageID: toolIdentity.assistantMessageID, callID: "call-question" } },
mode: "form",
fields: [],

View file

@ -1,5 +1,5 @@
import { describe, expect } from "bun:test"
import { DateTime, Effect, Layer, Schema, Tracer } from "effect"
import { DateTime, Effect, Fiber, Layer, Schema, Stream, Tracer } from "effect"
import {
ATTR_OPENCODE_LINK_TYPE,
ATTR_OPENCODE_SUBAGENT_AGENT_NAME,
@ -21,6 +21,7 @@ import { LocationServiceMap } from "@opencode-ai/core/location-service-map"
import { SessionV2 } from "@opencode-ai/core/session"
import { SessionEvent } from "@opencode-ai/core/session/event"
import { SessionExecution } from "@opencode-ai/core/session/execution"
import { SessionInput } from "@opencode-ai/core/session/input"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model"
import { SessionStore } from "@opencode-ai/core/session/store"
@ -306,6 +307,13 @@ describe("SubagentTool", () => {
return span
},
})
const events = yield* EventV2.Service
const admitted = yield* events.subscribe(SessionEvent.InputAdmitted).pipe(
Stream.filter((event) => event.data.sessionID === parent.id && event.data.input.type === "synthetic"),
Stream.take(1),
Stream.runCollect,
Effect.forkScoped({ startImmediately: true }),
)
const settled = yield* settleTool(registry, {
sessionID: parent.id,
@ -320,7 +328,10 @@ describe("SubagentTool", () => {
const childID = outputSessionID(settled.output?.structured)
expect(settled.output?.structured).toMatchObject({ status: "running" })
yield* Effect.yieldNow
const admission = Array.from(yield* Fiber.join(admitted))[0]
expect(admission?.data.input.data.text).toContain(`<subagent id="${childID}" state="completed"`)
const database = yield* Database.Service
yield* SessionInput.promoteSteers(database.db, events, parent.id)
const synthetic = (yield* sessions.context(parent.id)).filter((message) => message.type === "synthetic")
expect(synthetic).toHaveLength(1)
expect(synthetic[0]?.text).toContain(`<subagent id="${childID}" state="completed"`)