mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-15 22:26:40 +00:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6cb2f00d15 |
@@ -0,0 +1,5 @@
|
||||
---
|
||||
"@opencode-ai/core": patch
|
||||
---
|
||||
|
||||
Simplify interrupt continuation: the steer-scoped resume decision now lives in SessionExecution as a post-cleanup inbox check, and the run coordinator drops its continuation state machine. Wakes arriving during cancellation cleanup now restart a normal full drain, and interrupting an idle session with continue now resumes pending steering input.
|
||||
@@ -41,8 +41,8 @@
|
||||
"solid-js": "catalog:",
|
||||
"tree-sitter-bash": "0.25.0",
|
||||
"tree-sitter-powershell": "0.25.10",
|
||||
"uqr": "0.1.3",
|
||||
"web-tree-sitter": "0.25.10",
|
||||
"uqr": "0.1.3",
|
||||
"ws": "8.21.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
|
||||
@@ -21,7 +21,7 @@ export default function TermsOfService() {
|
||||
<section data-component="brand-content">
|
||||
<article data-component="terms-of-service">
|
||||
<h1>Terms of Use</h1>
|
||||
<p class="effective-date">Effective date: Aug 15, 2026</p>
|
||||
<p class="effective-date">Effective date: Mar 6, 2026</p>
|
||||
|
||||
<p>
|
||||
Welcome to OpenCode. Please read on to learn the rules and restrictions that govern your use of
|
||||
@@ -154,11 +154,6 @@ export default function TermsOfService() {
|
||||
is dangerous, harmful, fraudulent, deceptive, threatening, harassing, defamatory, obscene, or
|
||||
otherwise objectionable;
|
||||
</li>
|
||||
<li>
|
||||
creates, maintains, or uses accounts in bulk, or creates, maintains, or uses multiple accounts to
|
||||
circumvent usage limits, access restrictions, billing obligations, promotions, suspensions, or any
|
||||
other restriction or policy applicable to the Services;
|
||||
</li>
|
||||
<li>automatically or programmatically extracts data or Output (defined below);</li>
|
||||
<li>Represent that the Output was human-generated when it was not;</li>
|
||||
<li>
|
||||
|
||||
@@ -63,33 +63,6 @@ examples, but do not fetch it to determine the V2 configuration shape.
|
||||
See the [full configuration guide](https://opencode.ai/v2/docs/config) for
|
||||
every field, examples, config locations, and links to dedicated feature guides.
|
||||
|
||||
## [MCP servers](https://opencode.ai/v2/docs/mcp-servers)
|
||||
|
||||
Configure MCP servers under `mcp.servers`. Prefer the CLI because it preserves
|
||||
unrelated configuration. Use `--global` when the user asks to set up a service
|
||||
for themselves without limiting it to the current project; omit it when they
|
||||
explicitly want project-local configuration.
|
||||
|
||||
```sh
|
||||
opencode2 mcp add <name> --global --url <remote-url>
|
||||
opencode2 mcp list
|
||||
```
|
||||
|
||||
Remote servers use OAuth by default. If `mcp list` reports that a server needs
|
||||
authentication, run the OAuth flow and then verify the connection:
|
||||
|
||||
```sh
|
||||
opencode2 mcp auth <name>
|
||||
opencode2 mcp list
|
||||
```
|
||||
|
||||
The auth command prints an authorization URL, waits for the browser redirect,
|
||||
and stores credentials outside the OpenCode configuration. Do not ask for or
|
||||
store an API key when the server supports OAuth. Use header-based credentials
|
||||
only when OAuth is unavailable or the user explicitly requires them, and use an
|
||||
environment substitution such as `{env:MCP_API_KEY}` instead of writing a
|
||||
secret into configuration.
|
||||
|
||||
## [V1 to V2 migration](https://opencode.ai/v2/docs/migrate-v1)
|
||||
|
||||
For any request to migrate OpenCode configuration, agents, commands, skills,
|
||||
|
||||
@@ -137,13 +137,13 @@ export const layer = Layer.effect(
|
||||
return Service.of({
|
||||
active: coordinator.active,
|
||||
interrupt: (sessionID, options) =>
|
||||
coordinator.interrupt(
|
||||
sessionID,
|
||||
"user",
|
||||
options?.continue
|
||||
? { continue: { request: "steer", when: SessionInbox.has(db, sessionID, "steer") } }
|
||||
: undefined,
|
||||
),
|
||||
Effect.gen(function* () {
|
||||
yield* coordinator.interrupt(sessionID, "user")
|
||||
if (!options?.continue) return
|
||||
// Resume only steering input from the interrupted intent. Queued next-turn work
|
||||
// stays parked: a steer-scoped drain never promotes queue-delivery rows.
|
||||
if (yield* SessionInbox.has(db, sessionID, "steer")) yield* coordinator.wake(sessionID, "steer")
|
||||
}),
|
||||
resume: coordinator.run,
|
||||
wake: coordinator.wake,
|
||||
wakeActive: coordinator.wakeActive,
|
||||
|
||||
@@ -10,25 +10,19 @@ export interface Coordinator<Key, E, Reason = never> {
|
||||
/** Starts an execution while idle, or joins the active execution and returns its exit. */
|
||||
readonly run: (key: Key) => Effect.Effect<void, E>
|
||||
/** Rings the doorbell: an idle key starts an execution; an active one drains again before settling. */
|
||||
readonly wake: (key: Key, request?: Request) => Effect.Effect<void>
|
||||
/** Rings the current execution's doorbell with its existing request. Idle keys remain idle. */
|
||||
readonly wake: (key: Key, scope?: Promotable) => Effect.Effect<void>
|
||||
/** Rings the current execution's doorbell with its existing scope. Idle keys remain idle. */
|
||||
readonly wakeActive: (key: Key) => Effect.Effect<void>
|
||||
/** Stops the active execution, clears its doorbell, and waits for cleanup. No-op when idle. */
|
||||
readonly interrupt: (
|
||||
key: Key,
|
||||
reason?: Reason,
|
||||
options?: { readonly continue?: { readonly request: Request; readonly when: Effect.Effect<boolean> } },
|
||||
) => Effect.Effect<void>
|
||||
readonly interrupt: (key: Key, reason?: Reason) => Effect.Effect<void>
|
||||
/** Resolves once no execution is active for the key. Returns immediately when already idle and never starts work. */
|
||||
readonly awaitIdle: (key: Key) => Effect.Effect<void>
|
||||
}
|
||||
|
||||
export type Request = Promotable
|
||||
|
||||
/**
|
||||
* One execution is a busy period for one key: one fiber that drains from the first wake
|
||||
* until the key would stay idle. `pendingWake` is the doorbell: work recorded during the
|
||||
* execution rings it with its eligibility request, and the execution loop drains again
|
||||
* execution rings it with the scope that work needs, and the execution loop drains again
|
||||
* instead of ending. The doorbell closes the gap between a drain's last eligibility check
|
||||
* and the idle transition, since those cannot be one atomic step. `done` resolves joiners
|
||||
* with this execution's exit.
|
||||
@@ -36,15 +30,10 @@ export type Request = Promotable
|
||||
type Execution<E, Reason> = {
|
||||
readonly done: Deferred.Deferred<void, E>
|
||||
owner?: Fiber.Fiber<void>
|
||||
request: Request
|
||||
pendingWake?: Request
|
||||
scope: Promotable
|
||||
pendingWake?: Promotable
|
||||
stopping: boolean
|
||||
interruptionReason?: Reason
|
||||
continuation?: {
|
||||
readonly request: Request
|
||||
readonly when: Effect.Effect<boolean>
|
||||
signaled: boolean
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -59,7 +48,7 @@ type Execution<E, Reason> = {
|
||||
* ```
|
||||
*/
|
||||
export const make = <Key, E, Reason = never>(options: {
|
||||
readonly drain: (key: Key, force: boolean, request: Request) => Effect.Effect<void, E>
|
||||
readonly drain: (key: Key, force: boolean, scope: Promotable) => Effect.Effect<void, E>
|
||||
/** Runs once when a process-local busy period begins, before its first drain. */
|
||||
readonly started?: (key: Key) => Effect.Effect<void>
|
||||
/**
|
||||
@@ -73,11 +62,11 @@ export const make = <Key, E, Reason = never>(options: {
|
||||
const fork = yield* FiberSet.makeRuntime<never, void, never>()
|
||||
|
||||
const loop = (key: Key, execution: Execution<E, Reason>, force: boolean): Effect.Effect<void, E> =>
|
||||
Effect.suspend(() => options.drain(key, force, execution.request)).pipe(
|
||||
Effect.suspend(() => options.drain(key, force, execution.scope)).pipe(
|
||||
Effect.flatMap(() =>
|
||||
Effect.suspend(() => {
|
||||
if (execution.stopping || execution.pendingWake === undefined) return Effect.void
|
||||
execution.request = execution.pendingWake
|
||||
execution.scope = execution.pendingWake
|
||||
execution.pendingWake = undefined
|
||||
// Trampoline so drains that complete synchronously cannot grow the stack.
|
||||
return Effect.yieldNow.pipe(Effect.andThen(loop(key, execution, false)))
|
||||
@@ -85,10 +74,10 @@ export const make = <Key, E, Reason = never>(options: {
|
||||
),
|
||||
)
|
||||
|
||||
const start = (key: Key, force: boolean, request: Request) => {
|
||||
const start = (key: Key, force: boolean, scope: Promotable) => {
|
||||
const execution: Execution<E, Reason> = {
|
||||
done: Deferred.makeUnsafe<void, E>(),
|
||||
request,
|
||||
scope,
|
||||
stopping: false,
|
||||
}
|
||||
executions.set(key, execution)
|
||||
@@ -104,7 +93,7 @@ export const make = <Key, E, Reason = never>(options: {
|
||||
execution.owner = undefined
|
||||
}).pipe(Effect.andThen(options.settled?.(key, exit, execution.interruptionReason) ?? Effect.void)),
|
||||
),
|
||||
Effect.onExit((exit) => finish(key, execution, exit)),
|
||||
Effect.onExit((exit) => Effect.sync(() => settle(key, execution, exit))),
|
||||
Effect.exit,
|
||||
Effect.asVoid,
|
||||
),
|
||||
@@ -114,22 +103,12 @@ export const make = <Key, E, Reason = never>(options: {
|
||||
|
||||
// A doorbell that survives the execution loop (rung after the loop decided to end, or
|
||||
// during failure or interruption cleanup) starts a fresh execution for the remaining work.
|
||||
const settle = (key: Key, execution: Execution<E, Reason>, exit: Exit.Exit<void, E>, resume: boolean) => {
|
||||
if (resume && execution.continuation) start(key, false, execution.continuation.request)
|
||||
else if (execution.pendingWake) start(key, false, execution.pendingWake)
|
||||
const settle = (key: Key, execution: Execution<E, Reason>, exit: Exit.Exit<void, E>) => {
|
||||
if (execution.pendingWake) start(key, false, execution.pendingWake)
|
||||
else executions.delete(key)
|
||||
Deferred.doneUnsafe(execution.done, exit)
|
||||
}
|
||||
|
||||
const finish = (key: Key, execution: Execution<E, Reason>, exit: Exit.Exit<void, E>) => {
|
||||
if (!execution.continuation) return Effect.sync(() => settle(key, execution, exit, false))
|
||||
return execution.continuation.when.pipe(
|
||||
Effect.flatMap((ready) =>
|
||||
Effect.sync(() => settle(key, execution, exit, ready || execution.continuation?.signaled === true)),
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
const run = (key: Key): Effect.Effect<void, E> =>
|
||||
Effect.suspend(() => {
|
||||
const execution = executions.get(key)
|
||||
@@ -141,55 +120,32 @@ export const make = <Key, E, Reason = never>(options: {
|
||||
return Deferred.await(start(key, true, "input").done)
|
||||
})
|
||||
|
||||
const wake = (key: Key, request: Request = "input") =>
|
||||
const wake = (key: Key, scope: Promotable = "input") =>
|
||||
Effect.sync(() => {
|
||||
const execution = executions.get(key)
|
||||
if (execution !== undefined) {
|
||||
if (execution.stopping) {
|
||||
if (execution.continuation) execution.continuation.signaled = true
|
||||
else execution.continuation = { request, when: Effect.succeed(true), signaled: true }
|
||||
return
|
||||
}
|
||||
// Coalesced wakes keep the widest request: "input" subsumes "steer".
|
||||
execution.pendingWake = execution.pendingWake === "input" ? "input" : request
|
||||
// Coalesced wakes keep the widest scope: "input" subsumes "steer".
|
||||
execution.pendingWake = execution.pendingWake === "input" ? "input" : scope
|
||||
return
|
||||
}
|
||||
start(key, false, request)
|
||||
start(key, false, scope)
|
||||
})
|
||||
|
||||
const wakeActive = (key: Key) =>
|
||||
Effect.suspend(() => {
|
||||
const execution = executions.get(key)
|
||||
return execution ? wake(key, execution.request) : Effect.void
|
||||
return execution ? wake(key, execution.scope) : Effect.void
|
||||
})
|
||||
|
||||
const interrupt = (
|
||||
key: Key,
|
||||
reason?: Reason,
|
||||
options?: { readonly continue?: { readonly request: Request; readonly when: Effect.Effect<boolean> } },
|
||||
): Effect.Effect<void> =>
|
||||
const interrupt = (key: Key, reason?: Reason): Effect.Effect<void> =>
|
||||
Effect.suspend(() => {
|
||||
const execution = executions.get(key)
|
||||
if (execution === undefined) return Effect.void
|
||||
if (execution.stopping) {
|
||||
if (options?.continue)
|
||||
execution.continuation = {
|
||||
...options.continue,
|
||||
signaled: execution.continuation?.signaled ?? false,
|
||||
}
|
||||
return Deferred.await(execution.done).pipe(Effect.exit, Effect.asVoid)
|
||||
}
|
||||
if (execution.owner === undefined) {
|
||||
if (!options?.continue) return Effect.void
|
||||
execution.stopping = true
|
||||
execution.pendingWake = undefined
|
||||
execution.continuation = { ...options.continue, signaled: false }
|
||||
return Deferred.await(execution.done).pipe(Effect.exit, Effect.asVoid)
|
||||
}
|
||||
if (execution?.owner === undefined || execution.stopping) return Effect.void
|
||||
execution.stopping = true
|
||||
// Wakes recorded so far belong to the interrupted intent; the interrupt claims them.
|
||||
// Wakes arriving during cleanup are new admissions and restart normally at settle.
|
||||
execution.pendingWake = undefined
|
||||
execution.interruptionReason = reason
|
||||
if (options?.continue) execution.continuation = { ...options.continue, signaled: false }
|
||||
return Fiber.interrupt(execution.owner)
|
||||
})
|
||||
|
||||
|
||||
@@ -181,7 +181,9 @@ export const Plugin = {
|
||||
)
|
||||
|
||||
const background = input.background === true
|
||||
yield* context.progress({ sessionID: child.id, status: "running" })
|
||||
yield* context.progress({
|
||||
metadata: { sessionID: child.id, status: "running" },
|
||||
})
|
||||
|
||||
const run = Effect.gen(function* () {
|
||||
// The child session owns its agent/model (set at create); prompt only admits input.
|
||||
|
||||
@@ -14,8 +14,10 @@ import { SessionExecution } from "@opencode-ai/core/session/execution"
|
||||
import { SessionRestart } from "@opencode-ai/core/session/execution/restart"
|
||||
import { UserInterruptedError } from "@opencode-ai/core/session/error"
|
||||
import { SessionEvent } from "@opencode-ai/core/session/event"
|
||||
import { SessionInbox } from "@opencode-ai/core/session/inbox"
|
||||
import { SessionMessage } from "@opencode-ai/core/session/message"
|
||||
import { SessionRunner } from "@opencode-ai/core/session/runner/index"
|
||||
import { SessionTable } from "@opencode-ai/core/session/sql"
|
||||
import { SessionInboxTable, SessionTable } from "@opencode-ai/core/session/sql"
|
||||
import { SessionStore } from "@opencode-ai/core/session/store"
|
||||
import { Context, Deferred, Effect, Exit, Fiber, Layer, LayerMap, Scope } from "effect"
|
||||
import { eq } from "drizzle-orm"
|
||||
@@ -290,6 +292,113 @@ describe("SessionExecution lifecycle", () => {
|
||||
)
|
||||
})
|
||||
|
||||
describe("SessionExecution interrupt continuation", () => {
|
||||
it.effect("resumes only steering input after an interrupt with continue", () =>
|
||||
Effect.gen(function* () {
|
||||
const database = yield* Database.Service
|
||||
const sessionID = Session.ID.make("ses_continue_steer")
|
||||
yield* seedSessions(database, [sessionID])
|
||||
yield* seedInbox(database, sessionID, ["steer", "queue"])
|
||||
|
||||
const draining = yield* Deferred.make<void>()
|
||||
const drains: Array<{ force: boolean; promotable?: SessionInbox.Promotable }> = []
|
||||
const scope = yield* Scope.make()
|
||||
yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
|
||||
const context = yield* buildExecution(scope, (input) =>
|
||||
Effect.suspend(() => {
|
||||
drains.push({ force: input.force, promotable: input.promotable })
|
||||
if (drains.length > 1) return Effect.void
|
||||
return Deferred.succeed(draining, undefined).pipe(Effect.andThen(Effect.never))
|
||||
}),
|
||||
)
|
||||
const execution = Context.get(context, SessionExecution.Service)
|
||||
yield* execution.resume(sessionID).pipe(Effect.forkScoped)
|
||||
yield* Deferred.await(draining)
|
||||
|
||||
yield* execution.interrupt(sessionID, { continue: true })
|
||||
yield* execution.awaitIdle(sessionID)
|
||||
|
||||
// The successor drain is steer-scoped: queued next-turn work stays parked.
|
||||
expect(drains).toEqual([
|
||||
{ force: true, promotable: "input" },
|
||||
{ force: false, promotable: "steer" },
|
||||
])
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("stays parked after an interrupt with continue when only queued work remains", () =>
|
||||
Effect.gen(function* () {
|
||||
const database = yield* Database.Service
|
||||
const sessionID = Session.ID.make("ses_continue_parked")
|
||||
yield* seedSessions(database, [sessionID])
|
||||
yield* seedInbox(database, sessionID, ["queue"])
|
||||
|
||||
const draining = yield* Deferred.make<void>()
|
||||
const drains: Array<SessionInbox.Promotable | undefined> = []
|
||||
const scope = yield* Scope.make()
|
||||
yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
|
||||
const context = yield* buildExecution(scope, (input) =>
|
||||
Effect.suspend(() => {
|
||||
drains.push(input.promotable)
|
||||
return Deferred.succeed(draining, undefined).pipe(Effect.andThen(Effect.never))
|
||||
}),
|
||||
)
|
||||
const execution = Context.get(context, SessionExecution.Service)
|
||||
yield* execution.resume(sessionID).pipe(Effect.forkScoped)
|
||||
yield* Deferred.await(draining)
|
||||
|
||||
yield* execution.interrupt(sessionID, { continue: true })
|
||||
yield* execution.awaitIdle(sessionID)
|
||||
|
||||
expect(drains).toEqual(["input"])
|
||||
expect(yield* execution.active).toEqual(new Set())
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("an idle interrupt with continue resumes pending steers", () =>
|
||||
Effect.gen(function* () {
|
||||
const database = yield* Database.Service
|
||||
const sessionID = Session.ID.make("ses_continue_idle")
|
||||
yield* seedSessions(database, [sessionID])
|
||||
yield* seedInbox(database, sessionID, ["steer"])
|
||||
|
||||
const drains: Array<{ force: boolean; promotable?: SessionInbox.Promotable }> = []
|
||||
const scope = yield* Scope.make()
|
||||
yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
|
||||
const context = yield* buildExecution(scope, (input) =>
|
||||
Effect.sync(() => void drains.push({ force: input.force, promotable: input.promotable })),
|
||||
)
|
||||
const execution = Context.get(context, SessionExecution.Service)
|
||||
|
||||
yield* execution.interrupt(sessionID, { continue: true })
|
||||
yield* execution.awaitIdle(sessionID)
|
||||
|
||||
expect(drains).toEqual([{ force: false, promotable: "steer" }])
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
||||
function seedInbox(
|
||||
database: Database.Service["Service"],
|
||||
sessionID: Session.ID,
|
||||
deliveries: ReadonlyArray<SessionInbox.Delivery>,
|
||||
) {
|
||||
return database.db
|
||||
.insert(SessionInboxTable)
|
||||
.values(
|
||||
deliveries.map((delivery, index) => ({
|
||||
id: SessionMessage.ID.create(),
|
||||
session_id: sessionID,
|
||||
type: "compaction" as const,
|
||||
payload: {},
|
||||
delivery,
|
||||
enqueued_seq: index + 1,
|
||||
})),
|
||||
)
|
||||
.run()
|
||||
.pipe(Effect.orDie)
|
||||
}
|
||||
|
||||
function seedSessions(
|
||||
database: Database.Service["Service"],
|
||||
sessionIDs: ReadonlyArray<Session.ID>,
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { describe, expect } from "bun:test"
|
||||
import { Cause, Deferred, Effect, Exit, Fiber, Layer } from "effect"
|
||||
import { SessionInbox } from "@opencode-ai/core/session/inbox"
|
||||
import { SessionRunCoordinator } from "@opencode-ai/core/session/run-coordinator"
|
||||
import { testEffect } from "./lib/effect"
|
||||
|
||||
@@ -269,31 +270,24 @@ describe("SessionRunCoordinator", () => {
|
||||
),
|
||||
)
|
||||
|
||||
it.effect("replaces a settlement-window wake with a steer continuation", () =>
|
||||
it.effect("a settlement-window wake starts a fresh execution with its own scope", () =>
|
||||
Effect.scoped(
|
||||
Effect.gen(function* () {
|
||||
const settling = yield* Deferred.make<void>()
|
||||
const release = yield* Deferred.make<void>()
|
||||
const requests: SessionRunCoordinator.Request[] = []
|
||||
const scopes: SessionInbox.Promotable[] = []
|
||||
const coordinator = yield* SessionRunCoordinator.make({
|
||||
drain: (_key, _force, request) => Effect.sync(() => requests.push(request)),
|
||||
drain: (_key, _force, scope) => Effect.sync(() => scopes.push(scope)),
|
||||
settled: () => Deferred.succeed(settling, undefined).pipe(Effect.andThen(Deferred.await(release))),
|
||||
})
|
||||
|
||||
yield* coordinator.wake("session", "input")
|
||||
yield* coordinator.wake("session", "steer")
|
||||
yield* Deferred.await(settling)
|
||||
yield* coordinator.wake("session", "input")
|
||||
const interrupted = yield* coordinator
|
||||
.interrupt("session", undefined, {
|
||||
continue: { request: "steer", when: Effect.succeed(true) },
|
||||
})
|
||||
.pipe(Effect.forkChild)
|
||||
yield* Effect.yieldNow
|
||||
yield* Deferred.succeed(release, undefined)
|
||||
yield* Fiber.join(interrupted)
|
||||
yield* coordinator.awaitIdle("session")
|
||||
|
||||
expect(requests).toEqual(["input", "steer"])
|
||||
expect(scopes).toEqual(["steer", "input"])
|
||||
}),
|
||||
),
|
||||
)
|
||||
@@ -371,17 +365,17 @@ describe("SessionRunCoordinator", () => {
|
||||
),
|
||||
)
|
||||
|
||||
it.effect("coalesces drain requests with input taking precedence", () =>
|
||||
it.effect("coalesces drain scopes with input taking precedence", () =>
|
||||
Effect.scoped(
|
||||
Effect.gen(function* () {
|
||||
const firstStarted = yield* Deferred.make<void>()
|
||||
const release = yield* Deferred.make<void>()
|
||||
const requests: SessionRunCoordinator.Request[] = []
|
||||
const scopes: SessionInbox.Promotable[] = []
|
||||
const coordinator = yield* SessionRunCoordinator.make({
|
||||
drain: (_key, _force, request) =>
|
||||
drain: (_key, _force, scope) =>
|
||||
Effect.gen(function* () {
|
||||
requests.push(request)
|
||||
if (requests.length !== 1) return
|
||||
scopes.push(scope)
|
||||
if (scopes.length !== 1) return
|
||||
yield* Deferred.succeed(firstStarted, undefined)
|
||||
yield* Deferred.await(release)
|
||||
}),
|
||||
@@ -394,22 +388,22 @@ describe("SessionRunCoordinator", () => {
|
||||
yield* Deferred.succeed(release, undefined)
|
||||
yield* coordinator.awaitIdle("session")
|
||||
|
||||
expect(requests).toEqual(["steer", "input"])
|
||||
expect(scopes).toEqual(["steer", "input"])
|
||||
}),
|
||||
),
|
||||
)
|
||||
|
||||
it.effect("does not carry a completed input request into a steer drain", () =>
|
||||
it.effect("does not carry a completed input scope into a steer drain", () =>
|
||||
Effect.scoped(
|
||||
Effect.gen(function* () {
|
||||
const firstStarted = yield* Deferred.make<void>()
|
||||
const release = yield* Deferred.make<void>()
|
||||
const requests: SessionRunCoordinator.Request[] = []
|
||||
const scopes: SessionInbox.Promotable[] = []
|
||||
const coordinator = yield* SessionRunCoordinator.make({
|
||||
drain: (_key, _force, request) =>
|
||||
drain: (_key, _force, scope) =>
|
||||
Effect.gen(function* () {
|
||||
requests.push(request)
|
||||
if (requests.length !== 1) return
|
||||
scopes.push(scope)
|
||||
if (scopes.length !== 1) return
|
||||
yield* Deferred.succeed(firstStarted, undefined)
|
||||
yield* Deferred.await(release)
|
||||
}),
|
||||
@@ -421,7 +415,7 @@ describe("SessionRunCoordinator", () => {
|
||||
yield* Deferred.succeed(release, undefined)
|
||||
yield* coordinator.awaitIdle("session")
|
||||
|
||||
expect(requests).toEqual(["input", "steer"])
|
||||
expect(scopes).toEqual(["input", "steer"])
|
||||
}),
|
||||
),
|
||||
)
|
||||
@@ -431,12 +425,12 @@ describe("SessionRunCoordinator", () => {
|
||||
Effect.gen(function* () {
|
||||
const firstStarted = yield* Deferred.make<void>()
|
||||
const release = yield* Deferred.make<void>()
|
||||
const requests: SessionRunCoordinator.Request[] = []
|
||||
const scopes: SessionInbox.Promotable[] = []
|
||||
const coordinator = yield* SessionRunCoordinator.make({
|
||||
drain: (_key, _force, request) =>
|
||||
drain: (_key, _force, scope) =>
|
||||
Effect.gen(function* () {
|
||||
requests.push(request)
|
||||
if (requests.length !== 1) return
|
||||
scopes.push(scope)
|
||||
if (scopes.length !== 1) return
|
||||
yield* Deferred.succeed(firstStarted, undefined)
|
||||
yield* Deferred.await(release)
|
||||
}),
|
||||
@@ -449,61 +443,23 @@ describe("SessionRunCoordinator", () => {
|
||||
yield* Deferred.succeed(release, undefined)
|
||||
yield* coordinator.awaitIdle("session")
|
||||
|
||||
expect(requests).toEqual(["steer", "steer"])
|
||||
expect(scopes).toEqual(["steer", "steer"])
|
||||
}),
|
||||
),
|
||||
)
|
||||
|
||||
it.effect("coalesces overlapping interrupt continuations into one steer successor", () =>
|
||||
it.effect("a cleanup-era wake starts a successor with its own scope", () =>
|
||||
Effect.scoped(
|
||||
Effect.gen(function* () {
|
||||
const firstStarted = yield* Deferred.make<void>()
|
||||
const cleanupStarted = yield* Deferred.make<void>()
|
||||
const cleanupGate = yield* Deferred.make<void>()
|
||||
const requests: SessionRunCoordinator.Request[] = []
|
||||
const scopes: SessionInbox.Promotable[] = []
|
||||
const coordinator = yield* SessionRunCoordinator.make({
|
||||
drain: (_key, _force, request) =>
|
||||
drain: (_key, _force, scope) =>
|
||||
Effect.gen(function* () {
|
||||
requests.push(request)
|
||||
if (requests.length !== 1) return
|
||||
yield* Deferred.succeed(firstStarted, undefined)
|
||||
yield* Effect.never.pipe(
|
||||
Effect.onInterrupt(() =>
|
||||
Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))),
|
||||
),
|
||||
)
|
||||
}),
|
||||
})
|
||||
const continuation = { continue: { request: "steer" as const, when: Effect.succeed(false) } }
|
||||
|
||||
yield* coordinator.wake("session")
|
||||
yield* Deferred.await(firstStarted)
|
||||
const first = yield* coordinator.interrupt("session", undefined, continuation).pipe(Effect.forkChild)
|
||||
yield* Deferred.await(cleanupStarted)
|
||||
const second = yield* coordinator.interrupt("session", undefined, continuation).pipe(Effect.forkChild)
|
||||
yield* Effect.yieldNow
|
||||
yield* coordinator.wake("session", "input")
|
||||
yield* Deferred.succeed(cleanupGate, undefined)
|
||||
yield* Effect.all([Fiber.join(first), Fiber.join(second)])
|
||||
yield* coordinator.awaitIdle("session")
|
||||
|
||||
expect(requests).toEqual(["input", "steer"])
|
||||
}),
|
||||
),
|
||||
)
|
||||
|
||||
it.effect("a continuing interrupt replaces a cleanup-era input wake", () =>
|
||||
Effect.scoped(
|
||||
Effect.gen(function* () {
|
||||
const firstStarted = yield* Deferred.make<void>()
|
||||
const cleanupStarted = yield* Deferred.make<void>()
|
||||
const cleanupGate = yield* Deferred.make<void>()
|
||||
const requests: SessionRunCoordinator.Request[] = []
|
||||
const coordinator = yield* SessionRunCoordinator.make({
|
||||
drain: (_key, _force, request) =>
|
||||
Effect.gen(function* () {
|
||||
requests.push(request)
|
||||
if (requests.length !== 1) return
|
||||
scopes.push(scope)
|
||||
if (scopes.length !== 1) return
|
||||
yield* Deferred.succeed(firstStarted, undefined)
|
||||
yield* Effect.never.pipe(
|
||||
Effect.onInterrupt(() =>
|
||||
@@ -515,45 +471,16 @@ describe("SessionRunCoordinator", () => {
|
||||
|
||||
yield* coordinator.wake("session", "input")
|
||||
yield* Deferred.await(firstStarted)
|
||||
const plain = yield* coordinator.interrupt("session").pipe(Effect.forkChild)
|
||||
const interrupt = yield* coordinator.interrupt("session").pipe(Effect.forkChild)
|
||||
yield* Deferred.await(cleanupStarted)
|
||||
// A new admission during cancellation restarts normally: interruption only
|
||||
// claims the wakes recorded before it.
|
||||
yield* coordinator.wake("session", "input")
|
||||
const continuing = yield* coordinator
|
||||
.interrupt("session", undefined, {
|
||||
continue: { request: "steer", when: Effect.succeed(false) },
|
||||
})
|
||||
.pipe(Effect.forkChild)
|
||||
yield* Effect.yieldNow
|
||||
yield* Deferred.succeed(cleanupGate, undefined)
|
||||
yield* Effect.all([Fiber.join(plain), Fiber.join(continuing)])
|
||||
yield* Fiber.join(interrupt)
|
||||
yield* coordinator.awaitIdle("session")
|
||||
|
||||
expect(requests).toEqual(["input", "steer"])
|
||||
}),
|
||||
),
|
||||
)
|
||||
|
||||
it.effect("does not start a conditional continuation without eligible work", () =>
|
||||
Effect.scoped(
|
||||
Effect.gen(function* () {
|
||||
const started = yield* Deferred.make<void>()
|
||||
const requests: SessionRunCoordinator.Request[] = []
|
||||
const coordinator = yield* SessionRunCoordinator.make({
|
||||
drain: (_key, _force, request) =>
|
||||
Effect.sync(() => requests.push(request)).pipe(
|
||||
Effect.andThen(Deferred.succeed(started, undefined)),
|
||||
Effect.andThen(Effect.never),
|
||||
),
|
||||
})
|
||||
|
||||
yield* coordinator.wake("session")
|
||||
yield* Deferred.await(started)
|
||||
yield* coordinator.interrupt("session", undefined, {
|
||||
continue: { request: "steer", when: Effect.succeed(false) },
|
||||
})
|
||||
yield* coordinator.awaitIdle("session")
|
||||
|
||||
expect(requests).toEqual(["input"])
|
||||
expect(scopes).toEqual(["input", "input"])
|
||||
}),
|
||||
),
|
||||
)
|
||||
|
||||
@@ -298,7 +298,7 @@ describe("SubagentTool", () => {
|
||||
})
|
||||
const child = yield* sessions.get(outputSessionID(settled.metadata))
|
||||
expect(settled.metadata).toEqual({ sessionID: child.id, status: "completed" })
|
||||
expect(progress[0]).toEqual({ sessionID: child.id, status: "running" })
|
||||
expect(progress[0]?.metadata).toEqual({ sessionID: child.id, status: "running" })
|
||||
expect(child).toMatchObject({
|
||||
parentID: parent.id,
|
||||
location: parent.location,
|
||||
|
||||
Reference in New Issue
Block a user