diff --git a/apps/web/tests/preset-migration.snapshot.ts b/apps/web/tests/preset-migration.snapshot.ts index 1791e88639..3ca55f2f85 100644 --- a/apps/web/tests/preset-migration.snapshot.ts +++ b/apps/web/tests/preset-migration.snapshot.ts @@ -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() - const releaseSetup = Promise.withResolvers() - 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') diff --git a/packages/api/session-controller/README.i18n.yaml b/packages/api/session-controller/README.i18n.yaml index bf2da83432..6df1300fd1 100644 --- a/packages/api/session-controller/README.i18n.yaml +++ b/packages/api/session-controller/README.i18n.yaml @@ -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 diff --git a/packages/api/session-controller/README.md b/packages/api/session-controller/README.md index eb70ee3959..abbb52cd63 100644 --- a/packages/api/session-controller/README.md +++ b/packages/api/session-controller/README.md @@ -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. diff --git a/packages/api/session-controller/README.zh.md b/packages/api/session-controller/README.zh.md index eaae8243e5..58bdbbf48b 100644 --- a/packages/api/session-controller/README.zh.md +++ b/packages/api/session-controller/README.zh.md @@ -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 对进行中请求的变更重放会纠正其可用性,带序号的投影会覆盖缓存提示。 diff --git a/packages/api/session-controller/src/agent.ts b/packages/api/session-controller/src/agent.ts index 108ef3015b..4467fefc0f 100644 --- a/packages/api/session-controller/src/agent.ts +++ b/packages/api/session-controller/src/agent.ts @@ -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>() private readonly resumes = new Map>() private readonly claims = new Map>() private readonly imageAdmissionChains = new WeakMap>() @@ -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 | 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 { 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 { - 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) diff --git a/packages/api/session-controller/src/commands.ts b/packages/api/session-controller/src/commands.ts index 29871acc62..c91f6e5b02 100644 --- a/packages/api/session-controller/src/commands.ts +++ b/packages/api/session-controller/src/commands.ts @@ -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, diff --git a/packages/api/session-controller/tests/agent-initialization.host.spec.ts b/packages/api/session-controller/tests/agent-initialization.host.spec.ts deleted file mode 100644 index 5ecd4fd160..0000000000 --- a/packages/api/session-controller/tests/agent-initialization.host.spec.ts +++ /dev/null @@ -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 { - const read = Promise.withResolvers() - 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() - const release = Promise.withResolvers() - const setup = async (): Promise => { 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() - const release = Promise.withResolvers() - 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() - const release = Promise.withResolvers() - 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() - const release = Promise.withResolvers() - 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() - const conflict = Promise.withResolvers() - 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() - const release = Promise.withResolvers() - 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() - const finishComposition = Promise.withResolvers() - // 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() - const release = Promise.withResolvers() - 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() - const release = Promise.withResolvers() - 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 - } - }) -})