feat(opencode): task signals — structured returns, terse completion, sparse context, wake-on-message

packages/core/src/background-job.ts

@@ -6,6 +6,13 @@ import { makeGlobalNode } from "./effect/app-node"
 
 export type Status = "running" | "completed" | "error" | "cancelled"
 
+export type MessagePayload = {
+  childSessionID: string
+  parentSessionID: string
+  body: string
+  expectReply: boolean
+}
+
 export type Info = {
   id: string
   type: string
@@ -28,6 +35,7 @@ type Active = {
   output?: { sequence: number; text: string }
   tail: Deferred.Deferred<void>
   promoted: Deferred.Deferred<Info>
+  messaged: Deferred.Deferred<MessagePayload>
   onPromote?: Effect.Effect<void>
 }
 
@@ -48,6 +56,11 @@ type PromoteResult = {
   onPromote?: Effect.Effect<void>
 }
 
+type MessageResult = {
+  info?: Info
+  messaged?: Deferred.Deferred<MessagePayload>
+}
+
 type StartResult = { info: Info } | { info: Info; scope: Scope.Closeable; token: object }
 
 type ExtendResult =
@@ -92,6 +105,8 @@ export interface Interface {
   readonly extend: (input: ExtendInput) => Effect.Effect<boolean>
   readonly wait: (input: WaitInput) => Effect.Effect<WaitResult>
   readonly waitForPromotion: (id: string) => Effect.Effect<Info>
+  readonly message: (id: string, payload: MessagePayload) => Effect.Effect<Info | undefined>
+  readonly waitForMessage: (id: string) => Effect.Effect<MessagePayload>
   readonly promote: (id: string) => Effect.Effect<Info | undefined>
   readonly cancel: (id: string) => Effect.Effect<Info | undefined>
 }
@@ -206,6 +221,7 @@ export const make = Effect.gen(function* () {
         const started_at = yield* Clock.currentTimeMillis
         const done = yield* Deferred.make<Info>()
         const promoted = yield* Deferred.make<Info>()
+        const messaged = yield* Deferred.make<MessagePayload>()
         const tail = yield* Deferred.make<void>()
         const result = yield* SynchronizedRef.modifyEffect(
           state.jobs,
@@ -232,6 +248,7 @@ export const make = Effect.gen(function* () {
               next: 1,
               tail,
               promoted,
+              messaged,
               onPromote: input.onPromote,
             }
             return [{ info: snapshot(job), scope, token }, new Map(jobs).set(id, job)] as readonly [
@@ -307,6 +324,37 @@ export const make = Effect.gen(function* () {
     return yield* Deferred.await(job.promoted)
   })
 
+  const message: Interface["message"] = Effect.fn("BackgroundJob.message")(function* (id, payload) {
+    return yield* Effect.uninterruptible(
+      Effect.gen(function* () {
+        const result = yield* SynchronizedRef.modifyEffect(
+          state.jobs,
+          Effect.fnUntraced(function* (jobs) {
+            const job = jobs.get(id)
+            if (!job || job.info.status !== "running")
+              return [{} as MessageResult, jobs] as readonly [MessageResult, Map<string, Active>]
+            const next = {
+              ...job,
+              info: { ...job.info, metadata: { ...job.info.metadata, messaged: true } },
+            }
+            return [
+              { info: snapshot(next), messaged: job.messaged },
+              new Map(jobs).set(id, next),
+            ] as readonly [MessageResult, Map<string, Active>]
+          }),
+        )
+        if (result.info && result.messaged) yield* Deferred.succeed(result.messaged, payload).pipe(Effect.ignore)
+        return result.info
+      }),
+    )
+  })
+
+  const waitForMessage: Interface["waitForMessage"] = Effect.fn("BackgroundJob.waitForMessage")(function* (id) {
+    const job = (yield* SynchronizedRef.get(state.jobs)).get(id)
+    if (!job || job.info.status !== "running") return yield* Effect.never
+    return yield* Deferred.await(job.messaged)
+  })
+
   const promote: Interface["promote"] = Effect.fn("BackgroundJob.promote")(function* (id) {
     const result = yield* SynchronizedRef.modifyEffect(
       state.jobs,
@@ -357,7 +405,7 @@ export const make = Effect.gen(function* () {
     return result.info
   })
 
-  return Service.of({ list, get, start, extend, wait, waitForPromotion, promote, cancel })
+  return Service.of({ list, get, start, extend, wait, waitForPromotion, message, waitForMessage, promote, cancel })
 })
 
 const layer = Layer.effect(Service, make)

packages/core/src/cross-spawn-spawner.ts

@@ -6,6 +6,7 @@ import * as Effect from "effect/Effect"
 import * as Exit from "effect/Exit"
 import * as FileSystem from "effect/FileSystem"
 import * as Layer from "effect/Layer"
+import { LayerNode } from "./effect/layer-node"
 import * as Path from "effect/Path"
 import * as PlatformError from "effect/PlatformError"
 import * as Predicate from "effect/Predicate"
@@ -497,11 +498,12 @@ export const make = Effect.gen(function* () {
   return makeSpawner(spawnCommand)
 })
 
-const layer: Layer.Layer<ChildProcessSpawner, never, FileSystem.FileSystem | Path.Path> = Layer.effect(
+export const layer: Layer.Layer<ChildProcessSpawner, never, FileSystem.FileSystem | Path.Path> = Layer.effect(
   ChildProcessSpawner,
   make,
 )
 
 export const node = makeGlobalNode({ service: ChildProcessSpawner, layer, deps: [filesystem, path] })
+export const defaultLayer = layer
 
 export * as CrossSpawnSpawner from "./cross-spawn-spawner"

packages/core/src/database/database.ts

@@ -3,6 +3,7 @@ export * as Database from "./database"
 import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite"
 import { layer as sqliteLayer } from "#sqlite"
 import { Context, Effect, Layer } from "effect"
+import { LayerNode } from "../effect/layer-node"
 import { Global } from "../global"
 import { Flag } from "../flag/flag"
 import { isAbsolute, join } from "path"
@@ -19,7 +20,7 @@ export interface Interface {
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/v2/storage/Database") {}
 
-const layer = Layer.effect(
+export const layer = Layer.effect(
   Service,
   Effect.gen(function* () {
     const db = yield* makeDatabase

packages/core/src/database/migration/20260616095854_session_slug_unique.ts

@@ -0,0 +1,26 @@
+// NOTE (2026-06-16): session.slug is NOT unique and was never designed to be.
+// Slug.create() (packages/core/src/util/slug.ts) returns a random adjective-noun
+// pair from a small fixed word list, and a new session's slug starts as "" until
+// a title is generated — so with enough sessions the slug space saturates and
+// new inserts collide. An earlier version of this migration created a
+// `session_slug_unique` UNIQUE INDEX, which made Session.createNext throw on
+// every new session once the space filled (a real install with ~2800 sessions
+// could not create any new session). s2s cross-process addressing was reworked
+// to use the globally-unique session_id instead of the slug, so slug uniqueness
+// is not needed anywhere.
+//
+// This migration is now a self-healing no-op: it DROPS the bad index if a DB
+// applied the earlier version, and creates nothing. The id is preserved so the
+// migration journal stays consistent for DBs that already recorded it.
+
+import { Effect } from "effect"
+import type { DatabaseMigration } from "../migration"
+
+export default {
+  id: "20260616095854_session_slug_unique",
+  up(tx) {
+    return Effect.gen(function* () {
+      yield* tx.run(`DROP INDEX IF EXISTS session_slug_unique;`)
+    })
+  },
+} satisfies DatabaseMigration.Migration

packages/core/src/database/migration/20260616101412_s2s_tables.ts

@@ -0,0 +1,70 @@
+// Session-to-Session — Task 2 (store tables).
+//
+// Three tables backing the s2s store module in
+// `packages/opencode/src/s2s/store.ts`:
+//
+//   s2s_inbox   — durable cross-process mailbox. A target session ID drains
+//                rows from this table by atomically marking `drained_at`.
+//                The `drained_at IS NULL` guard inside the store's
+//                UPDATE…RETURNING claim is the cross-process double-claim
+//                protection: two concurrent drains racing on the same row
+//                will see exactly one claim succeed (Bun's SQLite WAL
+//                serializes writers, see Task 0's WAL sanity note in
+//                `20260616095854_session_slug_unique.ts`).
+//   s2s_token   — single-use invitation tokens issued by a session and
+//                consumed once by a joining session. `accepted_by` flips
+//                from NULL → session-id atomically; a NULL guard in the
+//                store's claim makes double-acceptance impossible.
+//   s2s_allow   — directional session-pair allowlist. Composite PK
+//                (session_id, allowed_session_id) makes "is X allowed to
+//                talk to Y?" a single SELECT; the PK is naturally
+//                directional so we don't need an extra index.
+//
+// The Drizzle schema mirror of these tables lives in
+// `packages/core/src/database/s2s.sql.ts` so the codegen pipeline in
+// `script/migration.ts` keeps `schema.gen.ts` in sync — without that
+// mirror, a fresh-in-memory database (e.g. test setup) would run
+// `schema.up(tx)` (the Drizzle-derived full schema) and never create the
+// s2s tables. The TypeScript migration is what runs on existing installs
+// when the `applyOnly` loop encounters the new id.
+
+import { Effect } from "effect"
+import type { DatabaseMigration } from "../migration"
+
+export default {
+  id: "20260616101412_s2s_tables",
+  up(tx) {
+    return Effect.gen(function* () {
+      yield* tx.run(`
+        CREATE TABLE s2s_inbox (
+          id TEXT PRIMARY KEY,
+          target_session_id TEXT NOT NULL,
+          from_session_id TEXT,
+          from_slug TEXT,
+          capsule TEXT NOT NULL,
+          drained_at INTEGER,
+          time_created INTEGER NOT NULL
+        );
+      `)
+      yield* tx.run(`CREATE INDEX s2s_inbox_target ON s2s_inbox (target_session_id, drained_at);`)
+      yield* tx.run(`
+        CREATE TABLE s2s_token (
+          token TEXT PRIMARY KEY,
+          inviter_session_id TEXT NOT NULL,
+          inviter_slug TEXT NOT NULL,
+          accepted_by TEXT,
+          accepted_at INTEGER,
+          created_at INTEGER NOT NULL
+        );
+      `)
+      yield* tx.run(`
+        CREATE TABLE s2s_allow (
+          session_id TEXT NOT NULL,
+          allowed_session_id TEXT NOT NULL,
+          established_at INTEGER NOT NULL,
+          PRIMARY KEY (session_id, allowed_session_id)
+        );
+      `)
+    })
+  },
+} satisfies DatabaseMigration.Migration

packages/core/src/database/migration/20260705045947_productive_masque.ts

@@ -0,0 +1,11 @@
+import { Effect } from "effect"
+import type { DatabaseMigration } from "../migration"
+
+export default {
+  id: "20260705045947_productive_masque",
+  up(tx) {
+    return Effect.gen(function* () {
+      yield* tx.run(`ALTER TABLE \`session\` ADD \`result\` text;`)
+    })
+  },
+} satisfies DatabaseMigration.Migration

packages/core/src/database/migration/20260705061319_silly_the_hood.ts

@@ -0,0 +1,11 @@
+import { Effect } from "effect"
+import type { DatabaseMigration } from "../migration"
+
+export default {
+  id: "20260705061319_silly_the_hood",
+  up(tx) {
+    return Effect.gen(function* () {
+      yield* tx.run(`ALTER TABLE \`session\` ADD \`context_mode\` text;`)
+    })
+  },
+} satisfies DatabaseMigration.Migration

packages/core/src/database/s2s.sql.ts

@@ -0,0 +1,48 @@
+// Drizzle schema declarations for the s2s_* tables.
+//
+// The s2s store (`packages/opencode/src/s2s/store.ts`) accesses these
+// tables via raw `sql\`\`` queries — it does NOT use the Drizzle query
+// builder — but the tables still need to appear in Drizzle's schema
+// graph so the codegen pipeline in `packages/core/script/migration.ts`
+// emits `CREATE TABLE` statements for them. Without a Drizzle definition,
+// a fresh in-memory database (e.g. test setup) ends up running
+// `schema.up(tx)` (which is just the Drizzle-derived full schema) and
+// never creates the s2s tables. The TypeScript migration
+// `20260616101412_s2s_tables` runs only on existing installs, where the
+// upgrade path is "find the new migration id in the registry, run its
+// `up`". Drizzle schema presence keeps both paths consistent.
+
+import { integer, sqliteTable, text, index, primaryKey } from "drizzle-orm/sqlite-core"
+
+export const S2SInboxTable = sqliteTable(
+  "s2s_inbox",
+  {
+    id: text().primaryKey(),
+    target_session_id: text().notNull(),
+    from_session_id: text(),
+    from_slug: text(),
+    capsule: text().notNull(),
+    drained_at: integer(),
+    time_created: integer().notNull(),
+  },
+  (table) => [index("s2s_inbox_target").on(table.target_session_id, table.drained_at)],
+)
+
+export const S2STokenTable = sqliteTable("s2s_token", {
+  token: text().primaryKey(),
+  inviter_session_id: text().notNull(),
+  inviter_slug: text().notNull(),
+  accepted_by: text(),
+  accepted_at: integer(),
+  created_at: integer().notNull(),
+})
+
+export const S2SAllowTable = sqliteTable(
+  "s2s_allow",
+  {
+    session_id: text().notNull(),
+    allowed_session_id: text().notNull(),
+    established_at: integer().notNull(),
+  },
+  (table) => [primaryKey({ columns: [table.session_id, table.allowed_session_id] })],
+)

packages/core/src/database/schema.gen.ts

@@ -23,6 +23,35 @@ export default {
           \`time_completed\` integer NOT NULL
         );
       `)
+      yield* tx.run(`
+        CREATE TABLE \`s2s_allow\` (
+          \`session_id\` text NOT NULL,
+          \`allowed_session_id\` text NOT NULL,
+          \`established_at\` integer NOT NULL,
+          CONSTRAINT \`s2s_allow_pk\` PRIMARY KEY(\`session_id\`, \`allowed_session_id\`)
+        );
+      `)
+      yield* tx.run(`
+        CREATE TABLE \`s2s_inbox\` (
+          \`id\` text PRIMARY KEY,
+          \`target_session_id\` text NOT NULL,
+          \`from_session_id\` text,
+          \`from_slug\` text,
+          \`capsule\` text NOT NULL,
+          \`drained_at\` integer,
+          \`time_created\` integer NOT NULL
+        );
+      `)
+      yield* tx.run(`
+        CREATE TABLE \`s2s_token\` (
+          \`token\` text PRIMARY KEY,
+          \`inviter_session_id\` text NOT NULL,
+          \`inviter_slug\` text NOT NULL,
+          \`accepted_by\` text,
+          \`accepted_at\` integer,
+          \`created_at\` integer NOT NULL
+        );
+      `)
       yield* tx.run(`
         CREATE TABLE \`account_state\` (
           \`id\` integer PRIMARY KEY,
@@ -195,6 +224,7 @@ export default {
           \`summary_files\` integer,
           \`summary_diffs\` text,
           \`metadata\` text,
+          \`result\` text,
           \`cost\` real DEFAULT 0 NOT NULL,
           \`tokens_input\` integer DEFAULT 0 NOT NULL,
           \`tokens_output\` integer DEFAULT 0 NOT NULL,
@@ -209,6 +239,7 @@ export default {
           \`time_updated\` integer NOT NULL,
           \`time_compacting\` integer,
           \`time_archived\` integer,
+          \`context_mode\` text,
           CONSTRAINT \`fk_session_project_id_project_id_fk\` FOREIGN KEY (\`project_id\`) REFERENCES \`project\`(\`id\`) ON DELETE CASCADE
         );
       `)
@@ -236,6 +267,7 @@ export default {
           CONSTRAINT \`fk_session_share_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
         );
       `)
+      yield* tx.run(`CREATE INDEX \`s2s_inbox_target\` ON \`s2s_inbox\` (\`target_session_id\`,\`drained_at\`);`)
       yield* tx.run(`CREATE UNIQUE INDEX \`event_aggregate_seq_idx\` ON \`event\` (\`aggregate_id\`,\`seq\`);`)
       yield* tx.run(`CREATE INDEX \`event_aggregate_type_seq_idx\` ON \`event\` (\`aggregate_id\`,\`type\`,\`seq\`);`)
       yield* tx.run(

packages/core/src/fs-util.ts

@@ -8,6 +8,7 @@ import type { PlatformError } from "effect/PlatformError"
 import { Glob } from "./util/glob"
 import { serviceUse } from "./effect/service-use"
 import { makeGlobalNode } from "./effect/app-node"
+import { LayerNode } from "./effect/layer-node"
 import { filesystem } from "./effect/app-node-platform"
 
 export namespace FSUtil {