mirror of
https://github.com/deepseek-ai/deepseek-harness.git
synced 2026-10-26 16:55:08 +00:00
Merge pull request #6966 from deepseek-harness/revert-6959-xtr/fix-web-agent-initialization
Revert "fix(session-controller): 在 Web 操作中等待本地 Agent 初始化"
This commit is contained in:
commit
2ae65c16b7
7 files changed
+7
-450
No files matched your search
@@ -5,7 +5,6 @@ import { dirname } from 'node:path'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
import { zstdCompressSync, zstdDecompressSync } from 'node:zlib'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { installModelSelection } from '@deepseek-ai/dsh-agent'
|
||||
import { SESSION_FORMAT_VERSION, SessionId } from '@deepseek-ai/dsh-session'
|
||||
import { generationLogPath } from '../../../packages/session/session-persistence-jsonl/src/format.ts'
|
||||
import { scanZstdFrames } from '../../../packages/session/session-persistence-jsonl/src/zstd.ts'
|
||||
@@ -18,9 +17,7 @@ const childFixturePath = fileURLToPath(new URL('../../../snapshots/web/preset-mi
|
||||
const fixturePath = fileURLToPath(new URL('../../../snapshots/web/preset-migration/session.v2.jsonl', import.meta.url))
|
||||
|
||||
describe.skipIf(webSnapshotMode() === 'record')('historical preset restoration through the Web Host', () => {
|
||||
it.each([
|
||||
['controller', false], ['controller', true], ['external', true],
|
||||
] as const)('resumes code as PTC through %s (selection events=%s)', async (activation, withSelections) => {
|
||||
it.each([false, true])('resumes code as PTC (selection events=%s)', async (withSelections) => {
|
||||
const scaffold = await launchWebScaffold()
|
||||
try {
|
||||
const id = SessionId('preset-migration')
|
||||
@@ -72,36 +69,7 @@ describe.skipIf(webSnapshotMode() === 'record')('historical preset restoration t
|
||||
}
|
||||
await expect(readFile(successor)).rejects.toMatchObject({ code: 'ENOENT' })
|
||||
|
||||
const resolved = activation === 'controller' ? await scaffold.ctx.sessionController.resolveAgent(id) : await (async () => {
|
||||
const setupEntered = Promise.withResolvers<undefined>()
|
||||
const releaseSetup = Promise.withResolvers<undefined>()
|
||||
const initialization = scaffold.ctx.agents.resume({
|
||||
resumeSessionId: id,
|
||||
setup: async (ctx) => {
|
||||
installModelSelection(ctx)
|
||||
await scaffold.ctx.agentPresets.mount(ctx, 'ptc')
|
||||
setupEntered.resolve(undefined)
|
||||
await releaseSetup.promise
|
||||
},
|
||||
})
|
||||
try {
|
||||
await Promise.race([setupEntered.promise, initialization])
|
||||
expect(scaffold.ctx.agents.get(id)).toBeUndefined()
|
||||
const lookup = scaffold.ctx.sessionController.resolveAgent(id)
|
||||
releaseSetup.resolve(undefined)
|
||||
const original = await initialization
|
||||
const result = await lookup
|
||||
expect({ sameAgent: 'agent' in result && result.agent === original.agent }).toMatchInlineSnapshot(`
|
||||
{
|
||||
"sameAgent": true,
|
||||
}
|
||||
`)
|
||||
return result
|
||||
} finally {
|
||||
releaseSetup.resolve(undefined)
|
||||
await initialization
|
||||
}
|
||||
})()
|
||||
const resolved = await scaffold.ctx.sessionController.resolveAgent(id)
|
||||
if ('error' in resolved) throw resolved.error
|
||||
expect(scaffold.ctx.agentPresets.composedPreset(resolved.agent.ctx)).toBe('ptc')
|
||||
expect(resolved.agent.session.header.agentPreset).toBe('ptc')
|
||||
|
||||
@@ -15,8 +15,8 @@
|
||||
en: 1fdf707322fb6996
|
||||
zh: 0204f7e9d4aa3cce
|
||||
/session-controller/use-this-package:
|
||||
en: 2bea5826e7c25c99
|
||||
zh: 40fbc767c15e4cb5
|
||||
en: aabd5be911e477f5
|
||||
zh: 2f7fdb44bc8777bc
|
||||
/session-controller/client-references:
|
||||
en: 41bdda1ac32df558
|
||||
zh: aa5d0e325924066b
|
||||
|
||||
@@ -37,10 +37,6 @@ Each endpoint states its activation policy. List reads only stored headers and p
|
||||
|
||||
Every Remote endpoint is access-controlled: a Peer whose principal is not the owner reaches an endpoint only when, on every resource the call touches, a binding of that principal kind's policy in [`dsh-api-remote-access`](../remote-access/README.md) grants a role listing the action ([Peer principals Agent Note](../../../.agents/notes/implemented/architecture/2026-10-12-peer-principal-and-remote-access.md)). The controller injects `remoteAccess` and registers the `session` resource kind with the default field `sessionId` through `typert.resources`. `session/page` and `session/follow` declare a `session` reader in `@Remote({ resources })` that reads `request.address.sessionId` for an ordinary Session address and `request.address.childSessionId` for a subagent address; whether access to the parent Session covers its children is the policy's rule. `session/projections` reads the Session resource from `request.sessionId`, so a reconnecting reader reloads only the projections of Sessions it may read. `session/list`, `session/search`, and `session/control` read no resource, so their calls touch the `global` resource and need a `global` role; for a non-owner caller, `session/list` returns only the rows, `session/search` only the hits, and `session/control` only the baseline projections and projection frames, whose Session `ctx.remoteAccess.allows()` admits for the same action, so that action also belongs to a `session` role; owner results are unfiltered. `session/prompt` builds the durable `user-rpc` message source in its Remote method and records `principal` only for a non-owner caller, read through `principalOf(this.ctx.invocation)`; a source without `principal` belongs to the owner.
|
||||
|
||||
Session history is readable before its Agent is ready; the summary reports `agentAvailable: false` during setup. Agent operations and named blank reuse wait for a local initialization whose Session was published while this controller was active, without acquiring its disposal ownership. Unpublished initializations, bare Sessions, and external writers retain their existing writer-conflict behavior.
|
||||
|
||||
If a joined initialization fails or is cancelled, waiters return `session/not-found` after the Session is released. Controller unload also ends its waiters with that error. These failures do not automatically restart initialization.
|
||||
|
||||
List rows carry `formatStatus` when available, so Clients can identify migration-required Sessions without requesting their projections.
|
||||
|
||||
Host list generation yields between complete rows only when its configured work slice is exhausted; cheap remaining work returns without a forced yield. Request cancellation is checked per row and after each yield and rejects without partial results. Each row is synchronous; the list does not promise one cross-Session snapshot. A Session already queued as cold remains a cached row if it attaches during a yield: Client in-flight mutation replay corrects its availability, and sequenced projections supersede cached hints.
|
||||
|
||||
@@ -37,10 +37,6 @@ Client journal 在发布 follow 快照、live entry 或历史页之前验证当
|
||||
|
||||
每个 Remote 端点都受访问控制:principal 不是 owner 的 Peer 只有在本次调用访问的每个资源上,该 principal kind 在 [`dsh-api-remote-access`](../remote-access/README.zh.md) 中的 policy 都有某个 binding 授予列出该 action 的 role 时,才能调用该端点([Peer 身份属性 Agent Note](../../../.agents/notes/implemented/architecture/2026-10-12-peer-principal-and-remote-access.zh.md))。控制器注入 `remoteAccess`,并经 `typert.resources` 以默认字段 `sessionId` 登记 `session` 资源种类。`session/page` 与 `session/follow` 在 `@Remote({ resources })` 中声明 `session` 读取函数:对普通 Session 地址读 `request.address.sessionId`,对 subagent 地址读 `request.address.childSessionId`;父会话的访问权是否覆盖其子会话由 policy 决定。`session/projections` 从 `request.sessionId` 读取会话资源,因此重连的读者只会重新读取它可读的会话的 projections。`session/list`、`session/search` 与 `session/control` 不读取资源,其调用访问 `global` 资源,需要 `global` role;对非 owner 调用方,`session/list` 只返回其 Session 被 `ctx.remoteAccess.allows()` 按同一 action 允许的行,`session/search` 只返回这类 Session 的命中,`session/control` 也只返回这类 Session 的 baseline projection 与 projection 帧,因此该 action 也要列在 `session` role 中;owner 的结果不过滤。`session/prompt` 在其 Remote 方法内构造持久的 `user-rpc` 消息 source,且只为非 owner 调用方记录经 `principalOf(this.ctx.invocation)` 读取的 `principal`;不带 `principal` 的 source 属于 owner。
|
||||
|
||||
Session 历史在 Agent 就绪前即可读取;setup 期间,摘要中的 `agentAvailable` 为 `false`。Agent 操作和命名空白会话复用会等待本地初始化,前提是其 Session 在此控制器活动期间发布;等待者不取得初始化所属 Agent 的销毁权限。尚未发布 Session 的初始化、裸 Session 和外部写入者仍遵循现有的写入冲突行为。
|
||||
|
||||
等待的初始化失败或被取消时,等待者在 Session 释放后收到 `session/not-found`。控制器卸载也会以同一错误结束其等待者。这些失败不会自动重启初始化。
|
||||
|
||||
列表行在状态可用时携带 `formatStatus`,Client 无需请求其投影即可识别需要迁移的 Session。
|
||||
|
||||
Host 仅在配置的工作时间片耗尽时,于完整条目之间让出事件循环;剩余工作较少时直接返回,不强制让出。每行及每次让出之后检查请求取消,取消时拒绝返回部分结果。每行同步计算;列表不承诺跨 Session 的统一快照。已经进入冷态队列的 Session 若在让出期间接入,仍返回缓存行:Client 对进行中请求的变更重放会纠正其可用性,带序号的投影会覆盖缓存提示。
|
||||
|
||||
@@ -8,7 +8,6 @@ import type {} from '@deepseek-ai/dsh-agent-default-model'
|
||||
import { installModelSelection } from '@deepseek-ai/dsh-agent'
|
||||
import type {} from '@deepseek-ai/dsh-agent-preset-registry'
|
||||
import { brandString } from '@deepseek-ai/dsh-brand'
|
||||
import { carrierKeyOf } from '@deepseek-ai/dsh-scope'
|
||||
import type { Session, SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { SessionInspection } from '@deepseek-ai/dsh-session-persistence'
|
||||
import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query'
|
||||
@@ -160,7 +159,6 @@ export async function inspectApiSession(
|
||||
|
||||
/** Owns every operation that may create, resume, or configure a Web Agent. */
|
||||
export class ApiSessionAgentController {
|
||||
private readonly initializing = new Map<Session, PromiseWithResolvers<Agent | undefined>>()
|
||||
private readonly resumes = new Map<SessionId, Promise<Agent>>()
|
||||
private readonly claims = new Map<SessionId, Promise<void>>()
|
||||
private readonly imageAdmissionChains = new WeakMap<Agent, Promise<void>>()
|
||||
@@ -173,25 +171,6 @@ export class ApiSessionAgentController {
|
||||
*/
|
||||
initialize(): void {
|
||||
const ctx = this.ctx
|
||||
const initializing = this.initializing
|
||||
ctx.on('session/created', function (session) {
|
||||
// AgentLoop publishes through its own Agent scope; bare Sessions and
|
||||
// Sessions created by another Agent do not announce their own readiness.
|
||||
const owner = carrierKeyOf(this) as Agent | undefined
|
||||
if (owner?.session === session) initializing.set(session, Promise.withResolvers())
|
||||
})
|
||||
ctx.on('agent/created', ({ agent }) => {
|
||||
initializing.get(agent.session)?.resolve(agent)
|
||||
initializing.delete(agent.session)
|
||||
})
|
||||
ctx.on('session/disposed', (session) => {
|
||||
initializing.get(session)?.resolve(undefined)
|
||||
initializing.delete(session)
|
||||
})
|
||||
ctx.effect(() => () => {
|
||||
for (const pending of initializing.values()) pending.resolve(undefined)
|
||||
initializing.clear()
|
||||
}, 'session-controller: Agent readiness')
|
||||
ctx.typert.lookups.configure('agent', async (sessionId: SessionId) => {
|
||||
const found = await this.resolveAgent(sessionId)
|
||||
if ('error' in found) throw found.error
|
||||
@@ -240,8 +219,7 @@ export class ApiSessionAgentController {
|
||||
|
||||
let resume = this.resumes.get(sessionId)
|
||||
if (resume === undefined) {
|
||||
resume = (this.pendingAgent(sessionId) ?? this.resume(sessionId, observation))
|
||||
.finally(() => { this.resumes.delete(sessionId) })
|
||||
resume = this.resume(sessionId, observation).finally(() => { this.resumes.delete(sessionId) })
|
||||
this.resumes.set(sessionId, resume)
|
||||
}
|
||||
try {
|
||||
@@ -260,7 +238,6 @@ export class ApiSessionAgentController {
|
||||
const raced = this.liveAgent(sessionId)
|
||||
if (raced !== undefined) return raced
|
||||
const racedSession = this.ctx.sessions.get(sessionId)
|
||||
if (racedSession !== undefined && this.initializing.has(racedSession)) return this.resolve(sessionId)
|
||||
if (racedSession !== undefined && hasApiSessionSubagentOwner(this.ctx, racedSession, undefined)) {
|
||||
return { error: apiSessionSubagentOwnershipError(sessionId) }
|
||||
}
|
||||
@@ -277,16 +254,6 @@ export class ApiSessionAgentController {
|
||||
}
|
||||
}
|
||||
|
||||
/** Observe Web readiness without acquiring the initializing Agent's disposal capability. */
|
||||
private pendingAgent(sessionId: SessionId): Promise<Agent> | undefined {
|
||||
const session = this.ctx.sessions.get(sessionId)
|
||||
const pending = session === undefined ? undefined : this.initializing.get(session)
|
||||
return pending?.promise.then((agent) => {
|
||||
if (agent === undefined) throw new ApiSessionNotFound(`session "${sessionId}" initialization did not complete`)
|
||||
return agent
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a Session for a new conversation, reusing the named blank Session
|
||||
* while it is still reusable. A reused Session runs the requested preset, or
|
||||
@@ -303,7 +270,6 @@ export class ApiSessionAgentController {
|
||||
* @throws ApiSessionSubagentOwnership when subagent routing owns the named Session.
|
||||
* @throws RemoteError `agent-preset/not-found` or `agent-preset/invalid` for the requested or default preset.
|
||||
* @throws ApiSessionPresetConflict when no roster is composed and the request names a preset for a reused Session.
|
||||
* @throws ApiSessionNotFound when a joined initialization ends without an Agent.
|
||||
*/
|
||||
async createSession(request: ApiSessionCreateRequest): Promise<ApiSessionCreated> {
|
||||
const named = request.sessionId
|
||||
@@ -385,8 +351,6 @@ export class ApiSessionAgentController {
|
||||
throw new ApiSessionSubagentOwnership(sessionId)
|
||||
}
|
||||
const composition = await this.composeAgent(this.presetForObservation(observation))
|
||||
const pending = this.pendingAgent(sessionId)
|
||||
if (pending !== undefined) return pending
|
||||
const published = this.ctx.sessions.get(sessionId)
|
||||
const live = this.ctx.agents.get(sessionId)
|
||||
if (published !== undefined && hasApiSessionSubagentOwner(this.ctx, published, live)) {
|
||||
@@ -412,7 +376,6 @@ export class ApiSessionAgentController {
|
||||
|
||||
/** @returns the reused Session, a new Session under `sessionId` when none is stored, or a new Session under a minted identity. */
|
||||
private async claim(sessionId: SessionId, request: ApiSessionCreateRequest): Promise<Agent> {
|
||||
await this.pendingAgent(sessionId)
|
||||
const attached = this.ctx.sessions.get(sessionId)
|
||||
const live = this.ctx.agents.get(sessionId)
|
||||
if (attached !== undefined && hasApiSessionSubagentOwner(this.ctx, attached, live)) {
|
||||
@@ -456,7 +419,6 @@ export class ApiSessionAgentController {
|
||||
if (error instanceof ApiSessionStartedSinceObserved) return this.createMinted(request.cwd, composition)
|
||||
if (!(error instanceof Error) || error.name !== 'SessionAlreadyOwnedError') throw error
|
||||
// Another resume in this process may hold the writer for the Agent it published.
|
||||
await this.pendingAgent(sessionId)
|
||||
return this.ctx.agents.get(sessionId) === undefined
|
||||
? this.createMinted(request.cwd, composition)
|
||||
: this.claim(sessionId, request)
|
||||
|
||||
@@ -132,7 +132,7 @@ export class SessionCommandController {
|
||||
&& (workspace === undefined || workspace.sessionIds.includes(sessionId)),
|
||||
})
|
||||
} catch (error) {
|
||||
this.rejectCreation(error, request.sessionId)
|
||||
this.rejectCreation(error)
|
||||
}
|
||||
const sessionId = created.agent.id
|
||||
if (workspace !== undefined) {
|
||||
@@ -513,11 +513,8 @@ export class SessionCommandController {
|
||||
return found.agent
|
||||
}
|
||||
|
||||
private rejectCreation(error: unknown, sessionId: SessionId | undefined): never {
|
||||
private rejectCreation(error: unknown): never {
|
||||
if (remoteErrorOf(error) !== undefined) throw error
|
||||
if (sessionId !== undefined && error instanceof ApiSessionNotFound) {
|
||||
throw new RemoteError('session/not-found', error.message, { sessionId })
|
||||
}
|
||||
if (error instanceof ApiSessionPresetConflict) {
|
||||
throw new RemoteError('agent-preset/conflict', error.message, {
|
||||
sessionId: error.sessionId,
|
||||
|
||||
@@ -1,362 +0,0 @@
|
||||
/** Web operations join the original Agent setup while history remains readable. */
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { Context } from '@deepseek-ai/cordis'
|
||||
import { mountAgentLoopTestDependencies, mountAgentLoopTestHarness } from '@deepseek-ai/dsh-agent-loop-testkit'
|
||||
import { SessionId, SessionSeq, type SessionEvent } from '@deepseek-ai/dsh-session'
|
||||
import * as JsonlPersistence from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
import SessionTitle from '@deepseek-ai/dsh-session-title'
|
||||
import { SessionAlreadyOwnedError } from '@deepseek-ai/dsh-session-persistence'
|
||||
import { describe, expect, it, onTestFinished, vi } from 'vitest'
|
||||
import { ApiSessionAgentController } from '../src/agent.ts'
|
||||
import type { SessionRequestId } from '../src/types.ts'
|
||||
import { createSessionTestController, createSessionTestRemote } from './test-remote.ts'
|
||||
|
||||
const seed: SessionEvent[] = [
|
||||
{ type: 'turn/start', seq: SessionSeq(0), time: 1, data: { turn: 1 } },
|
||||
{ type: 'turn/end', seq: SessionSeq(1), time: 2, data: { turn: 1, reason: { kind: 'completed' } } },
|
||||
]
|
||||
|
||||
async function harness() {
|
||||
const root = await mkdtemp(join(tmpdir(), 'dsh-web-readiness-'))
|
||||
const ctx = new Context()
|
||||
onTestFinished(async () => {
|
||||
await ctx.fiber.dispose()
|
||||
await rm(root, { recursive: true, force: true })
|
||||
})
|
||||
await mountAgentLoopTestDependencies(ctx)
|
||||
await ctx.plugin(JsonlPersistence, { root })
|
||||
await mountAgentLoopTestHarness(ctx)
|
||||
await ctx.plugin(SessionTitle, { fallbackMaxWords: 5, fallbackMaxBytes: 40, maxTitleBytes: 80 })
|
||||
ctx.provide('workspaceRegistry', { get: () => undefined, list: () => [], archivedSessionIds: [] } as never)
|
||||
ctx.on('agent/pre-step', async () => ({ kind: 'reject' }))
|
||||
const defaults = { defaultModelSelection: () => ({ provider: 'fixture', model: 'fixture' }), cwd: root }
|
||||
const api = ctx.plugin({
|
||||
inject: ['agents', 'sessions', 'sessionProjections', 'llm', 'workspaceRegistry', 'sessionPersistence', 'sessionTitle'],
|
||||
apply() {},
|
||||
})
|
||||
await api
|
||||
const controller = await createSessionTestController(api.ctx, defaults)
|
||||
const remote = await createSessionTestRemote(api.ctx, defaults)
|
||||
return { ctx, root, controller, remote, api }
|
||||
}
|
||||
|
||||
function nextSessionRead(ctx: Context, sessionId: SessionId): Promise<undefined> {
|
||||
const read = Promise.withResolvers<undefined>()
|
||||
const get = ctx.sessions.get.bind(ctx.sessions)
|
||||
const spy = vi.spyOn(ctx.sessions, 'get').mockImplementation((id) => {
|
||||
const session = get(id)
|
||||
if (id === sessionId) read.resolve(undefined)
|
||||
return session
|
||||
})
|
||||
onTestFinished(() => { spy.mockRestore() })
|
||||
return read.promise
|
||||
}
|
||||
|
||||
describe('Web Agent initialization', () => {
|
||||
it.each(['create', 'resume'] as const)('keeps %s history readable and joins setup for mutations', async (mode) => {
|
||||
const { ctx, root, controller, remote } = await harness()
|
||||
const sessionId = SessionId('initializing')
|
||||
if (mode === 'resume') {
|
||||
const previous = await ctx.agents.create({ sessionId, meta: { cwd: root }, seed })
|
||||
await previous.dispose()
|
||||
}
|
||||
const entered = Promise.withResolvers<undefined>()
|
||||
const release = Promise.withResolvers<undefined>()
|
||||
const setup = async (): Promise<void> => { entered.resolve(undefined); await release.promise }
|
||||
const creation = mode === 'create'
|
||||
? ctx.agents.create({ sessionId, meta: { cwd: root }, seed, setup })
|
||||
: ctx.agents.resume({ resumeSessionId: sessionId, setup })
|
||||
onTestFinished(async () => { release.resolve(undefined); await creation })
|
||||
await entered.promise
|
||||
const resume = vi.spyOn(ctx.agents, 'resume')
|
||||
const abort = new AbortController()
|
||||
const iterator = controller.follow({ address: { kind: 'session', sessionId } }, abort.signal)
|
||||
[Symbol.asyncIterator]()
|
||||
try {
|
||||
expect(ctx.sessions.get(sessionId)).toBeDefined()
|
||||
expect(ctx.agents.get(sessionId)).toBeUndefined()
|
||||
expect((await controller.list({}, abort.signal)).items.find(item => item.sessionId === sessionId))
|
||||
.toMatchObject({ blank: false, agentAvailable: false })
|
||||
const snapshot = await iterator.next()
|
||||
if (snapshot.done || snapshot.value.type !== 'snapshot') throw new Error('missing history snapshot')
|
||||
expect(snapshot.value.records.slice(0, 2).map(record => record.event)).toEqual(seed)
|
||||
const resolved = controller.resolveAgent(sessionId)
|
||||
const renamed = remote.rename({ sessionId, title: 'Ready title' })
|
||||
const prompted = remote.prompt({
|
||||
requestId: 'ready-prompt' as SessionRequestId, sessionId, mode: 'queue',
|
||||
content: [{ type: 'text', text: 'hello' }], clientTimeZone: 'UTC',
|
||||
})
|
||||
release.resolve(undefined)
|
||||
const handle = await creation
|
||||
const resolution = await resolved
|
||||
expect('agent' in resolution && resolution.agent).toBe(handle.agent)
|
||||
await expect(renamed).resolves.toMatchObject({ ok: true, value: { title: 'Ready title' } })
|
||||
await expect(prompted).resolves.toMatchObject({ ok: true })
|
||||
expect(resume).not.toHaveBeenCalled()
|
||||
await ctx.sessions.flush(handle.agent.session)
|
||||
await using stored = await ctx.sessionPersistence.open(sessionId, 'read')
|
||||
expect((await stored.read()).events.findLast(event => event.type === 'session/title')?.data)
|
||||
.toMatchObject({ title: 'Ready title' })
|
||||
} finally {
|
||||
release.resolve(undefined)
|
||||
abort.abort()
|
||||
await iterator.return?.()
|
||||
resume.mockRestore()
|
||||
}
|
||||
})
|
||||
|
||||
it('reuses a named blank after its original setup completes', async () => {
|
||||
const { ctx, root, remote } = await harness()
|
||||
const sessionId = SessionId('blank-initializing')
|
||||
const entered = Promise.withResolvers<undefined>()
|
||||
const release = Promise.withResolvers<undefined>()
|
||||
const creation = ctx.agents.create({
|
||||
sessionId, meta: { cwd: root },
|
||||
setup: async () => { entered.resolve(undefined); await release.promise },
|
||||
})
|
||||
const resume = vi.spyOn(ctx.agents, 'resume')
|
||||
try {
|
||||
await entered.promise
|
||||
const reused = remote.create({ sessionId, cwd: root })
|
||||
release.resolve(undefined)
|
||||
await creation
|
||||
const result = await reused
|
||||
if (!result.ok) throw result.error
|
||||
expect(result.value.sessionId).toBe(sessionId)
|
||||
expect(ctx.agents.list()).toHaveLength(1)
|
||||
expect(resume).not.toHaveBeenCalled()
|
||||
} finally {
|
||||
release.resolve(undefined)
|
||||
await creation
|
||||
resume.mockRestore()
|
||||
}
|
||||
})
|
||||
|
||||
it.each(['failure', 'cancel'] as const)('ends waiters after initialization %s releases its writer', async (outcome) => {
|
||||
const { ctx, root, controller } = await harness()
|
||||
const sessionId = SessionId('failed-initializing')
|
||||
const entered = Promise.withResolvers<undefined>()
|
||||
const release = Promise.withResolvers<undefined>()
|
||||
const cancellation = new AbortController()
|
||||
const creation = ctx.agents.create({
|
||||
sessionId, meta: { cwd: root }, seed, signal: cancellation.signal,
|
||||
setup: async () => { entered.resolve(undefined); await release.promise; throw new Error('setup failed') },
|
||||
})
|
||||
const failed = expect(creation).rejects.toThrow()
|
||||
try {
|
||||
await entered.promise
|
||||
const resolution = controller.resolveAgent(sessionId)
|
||||
if (outcome === 'cancel') cancellation.abort(new Error('cancelled'))
|
||||
release.resolve(undefined)
|
||||
await failed
|
||||
await expect(resolution).resolves.toMatchObject({ error: { code: 'session/not-found' } })
|
||||
expect(ctx.agents.get(sessionId)).toBeUndefined()
|
||||
await using writer = await ctx.sessionPersistence.open(sessionId, 'write')
|
||||
expect(writer.id).toBe(sessionId)
|
||||
} finally {
|
||||
release.resolve(undefined)
|
||||
await failed
|
||||
}
|
||||
})
|
||||
|
||||
it.each(['failure', 'cancel'] as const)('reports named blank initialization %s without starting another Agent', async (outcome) => {
|
||||
const { ctx, root, remote } = await harness()
|
||||
const sessionId = SessionId('failed-blank')
|
||||
const entered = Promise.withResolvers<undefined>()
|
||||
const release = Promise.withResolvers<undefined>()
|
||||
const cancellation = new AbortController()
|
||||
const creation = ctx.agents.create({
|
||||
sessionId, meta: { cwd: root }, signal: cancellation.signal,
|
||||
setup: async () => {
|
||||
entered.resolve(undefined)
|
||||
await release.promise
|
||||
if (outcome === 'failure') throw new Error('setup failed')
|
||||
},
|
||||
})
|
||||
const failed = expect(creation).rejects.toThrow()
|
||||
const create = vi.spyOn(ctx.agents, 'create')
|
||||
try {
|
||||
await entered.promise
|
||||
const published = ctx.sessions.get(sessionId)
|
||||
if (published === undefined) throw new Error('initializing Session was not published')
|
||||
const waiting = nextSessionRead(ctx, sessionId)
|
||||
const reused = remote.create({ sessionId, cwd: root })
|
||||
await waiting
|
||||
if (outcome === 'cancel') cancellation.abort(new Error('cancelled'))
|
||||
release.resolve(undefined)
|
||||
await failed
|
||||
await expect(reused).resolves.toMatchObject({ ok: false, error: { code: 'session/not-found', details: { sessionId } } })
|
||||
expect(create).not.toHaveBeenCalled()
|
||||
expect(ctx.agents.list()).toHaveLength(0)
|
||||
await using writer = await ctx.sessionPersistence.create(published.header)
|
||||
expect(writer.id).toBe(sessionId)
|
||||
} finally {
|
||||
release.resolve(undefined)
|
||||
await failed
|
||||
create.mockRestore()
|
||||
}
|
||||
})
|
||||
|
||||
it.each([
|
||||
['resolve', 'success'], ['create', 'success'], ['create', 'failure'], ['create', 'cancel'],
|
||||
] as const)('joins the external writer after %s races its acquisition (%s)', async (operation, outcome) => {
|
||||
const { ctx, root, controller, remote } = await harness()
|
||||
const sessionId = SessionId('writer-race')
|
||||
const previous = await ctx.agents.create({ sessionId, meta: { cwd: root }, seed: [
|
||||
{ type: 'session/title', seq: SessionSeq(0), time: 1, data: { title: 'Blank', messageSeqs: [], source: { kind: 'user' } } },
|
||||
] })
|
||||
await previous.dispose()
|
||||
const acquiring = Promise.withResolvers<undefined>()
|
||||
const conflict = Promise.withResolvers<undefined>()
|
||||
const resume = vi.spyOn(ctx.agents, 'resume').mockImplementationOnce(async () => {
|
||||
acquiring.resolve(undefined)
|
||||
await conflict.promise
|
||||
throw new SessionAlreadyOwnedError(sessionId)
|
||||
})
|
||||
const request = operation === 'resolve'
|
||||
? controller.resolveAgent(sessionId).then(result => 'agent' in result
|
||||
? { sessionId: result.agent.id, agent: result.agent } : result)
|
||||
: remote.create({ sessionId, cwd: root }).then(result => result.ok ? result.value : { error: result.error })
|
||||
const entered = Promise.withResolvers<undefined>()
|
||||
const release = Promise.withResolvers<undefined>()
|
||||
const cancellation = new AbortController()
|
||||
onTestFinished(async () => {
|
||||
conflict.resolve(undefined)
|
||||
release.resolve(undefined)
|
||||
await request
|
||||
resume.mockRestore()
|
||||
})
|
||||
await Promise.race([acquiring.promise, request.then((result) => {
|
||||
if ('error' in result) throw result.error
|
||||
throw new Error('request settled before write acquisition')
|
||||
})])
|
||||
const initialization = ctx.agents.resume({
|
||||
resumeSessionId: sessionId, signal: cancellation.signal,
|
||||
setup: async () => {
|
||||
entered.resolve(undefined)
|
||||
await release.promise
|
||||
if (outcome === 'failure') throw new Error('setup failed')
|
||||
},
|
||||
})
|
||||
const settled = initialization.then(handle => ({ handle }), (error: unknown) => ({ error }))
|
||||
onTestFinished(async () => {
|
||||
release.resolve(undefined)
|
||||
await settled
|
||||
})
|
||||
await Promise.race([entered.promise, initialization])
|
||||
const waiting = nextSessionRead(ctx, sessionId)
|
||||
conflict.resolve(undefined)
|
||||
await waiting
|
||||
if (outcome === 'cancel') cancellation.abort(new Error('cancelled'))
|
||||
release.resolve(undefined)
|
||||
const original = await settled
|
||||
const result = await request
|
||||
if (outcome === 'success') {
|
||||
if ('error' in original) throw original.error
|
||||
expect(result).toMatchObject({ sessionId })
|
||||
if ('agent' in result) expect(result.agent).toBe(original.handle.agent)
|
||||
expect(ctx.agents.get(sessionId)).toBe(original.handle.agent)
|
||||
} else {
|
||||
expect(original).toHaveProperty('error')
|
||||
expect(result).toMatchObject({ error: { code: 'session/not-found', details: { sessionId } } })
|
||||
expect(ctx.agents.list()).toHaveLength(0)
|
||||
await using writer = await ctx.sessionPersistence.open(sessionId, 'write')
|
||||
expect(writer.id).toBe(sessionId)
|
||||
}
|
||||
expect(resume).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('joins initialization that starts during preset resolution without another resume', async () => {
|
||||
const { ctx, root, controller } = await harness()
|
||||
const sessionId = SessionId('composition-race')
|
||||
const previous = await ctx.agents.create({ sessionId, meta: { cwd: root }, seed })
|
||||
await previous.dispose()
|
||||
const composing = Promise.withResolvers<undefined>()
|
||||
const finishComposition = Promise.withResolvers<undefined>()
|
||||
// oxlint-disable-next-line typescript/unbound-method -- Invoked with the intercepted controller through call().
|
||||
const compose = ApiSessionAgentController.prototype.composeAgent
|
||||
const composition = vi.spyOn(ApiSessionAgentController.prototype, 'composeAgent')
|
||||
.mockImplementationOnce(async function (this: ApiSessionAgentController, preset) {
|
||||
composing.resolve(undefined)
|
||||
await finishComposition.promise
|
||||
return compose.call(this, preset)
|
||||
})
|
||||
const resolving = controller.resolveAgent(sessionId)
|
||||
const entered = Promise.withResolvers<undefined>()
|
||||
const release = Promise.withResolvers<undefined>()
|
||||
onTestFinished(async () => {
|
||||
finishComposition.resolve(undefined)
|
||||
release.resolve(undefined)
|
||||
await resolving
|
||||
composition.mockRestore()
|
||||
})
|
||||
await composing.promise
|
||||
const initialization = ctx.agents.resume({
|
||||
resumeSessionId: sessionId,
|
||||
setup: async () => { entered.resolve(undefined); await release.promise },
|
||||
})
|
||||
const resume = vi.spyOn(ctx.agents, 'resume')
|
||||
try {
|
||||
await Promise.race([entered.promise, initialization])
|
||||
const waiting = nextSessionRead(ctx, sessionId)
|
||||
finishComposition.resolve(undefined)
|
||||
await waiting
|
||||
release.resolve(undefined)
|
||||
const original = await initialization
|
||||
const result = await resolving
|
||||
expect('agent' in result && result.agent).toBe(original.agent)
|
||||
expect(resume).not.toHaveBeenCalled()
|
||||
} finally {
|
||||
finishComposition.resolve(undefined)
|
||||
release.resolve(undefined)
|
||||
await initialization
|
||||
await resolving
|
||||
resume.mockRestore()
|
||||
}
|
||||
})
|
||||
|
||||
it('does not wait for a bare Session whose writer belongs to another owner', async () => {
|
||||
const { ctx, root, controller } = await harness()
|
||||
const sessionId = SessionId('bare-session')
|
||||
const session = ctx.sessions.create(sessionId, { meta: { cwd: root } })
|
||||
await using writer = await ctx.sessionPersistence.create(session.header)
|
||||
expect(writer.id).toBe(sessionId)
|
||||
await expect(controller.resolveAgent(sessionId)).resolves.toMatchObject({ error: { code: 'session/writer-held' } })
|
||||
})
|
||||
|
||||
it('releases Controller waiters without disposing the original Agent owner', async () => {
|
||||
const { ctx, root, controller, api } = await harness()
|
||||
const sessionId = SessionId('independently-owned')
|
||||
const previous = await ctx.agents.create({ sessionId, meta: { cwd: root }, seed })
|
||||
await previous.dispose()
|
||||
const abort = new AbortController()
|
||||
const iterator = controller.follow({ address: { kind: 'session', sessionId } }, abort.signal)[Symbol.asyncIterator]()
|
||||
await iterator.next()
|
||||
const entered = Promise.withResolvers<undefined>()
|
||||
const release = Promise.withResolvers<undefined>()
|
||||
const creation = ctx.agents.resume({
|
||||
resumeSessionId: sessionId,
|
||||
setup: async () => { entered.resolve(undefined); await release.promise },
|
||||
})
|
||||
try {
|
||||
await entered.promise
|
||||
// Continuing the cold follow starts its background promotion while the original resume owns setup.
|
||||
const following = iterator.next()
|
||||
const resolving = controller.resolveAgent(sessionId)
|
||||
await api.dispose()
|
||||
await following
|
||||
await expect(resolving).resolves.toMatchObject({ error: { code: 'session/not-found' } })
|
||||
expect(ctx.sessions.get(sessionId)).toBeDefined()
|
||||
release.resolve(undefined)
|
||||
const handle = await creation
|
||||
expect(ctx.agents.get(sessionId)).toBe(handle.agent)
|
||||
} finally {
|
||||
release.resolve(undefined)
|
||||
abort.abort()
|
||||
await iterator.return?.()
|
||||
await creation
|
||||
}
|
||||
})
|
||||
})
|
||||
Reference in new issue
Block a user