diff --git a/packages/api/session-controller/README.i18n.yaml b/packages/api/session-controller/README.i18n.yaml index 3d1a5b274d..8ae33a613e 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: d8c0e2b2e45fa75d - zh: 025b50be4204580c + en: 807a42e6f5e5fb43 + zh: 23cf50e120c86c26 /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 ed01827a70..11f04d6d6e 100644 --- a/packages/api/session-controller/README.md +++ b/packages/api/session-controller/README.md @@ -33,6 +33,8 @@ The Client journal validates current Session event envelopes before publishing f Each endpoint states its activation policy. List reads only stored headers and projection-cache rows: it never calls per-session stat or opens a cold Session body. A current-format cache identity may supply every list hint; a lifecycle-matching predecessor cache may supply only its version-compatible title as a stale display fact, never as an authoritative fold seed. Search, attachment, history pages, log following, skill discovery, and workspace-path opening can inspect persistence without activating an Agent; `canOpenWorkspacePath()` reports native-opening availability without addressing a Session. Cancellation requires live state; queue mutation, model, rename, prompt, and file-reference operations may resolve or resume an ordinary Session. Prompt rejects content with neither non-whitespace text nor an attachment before resolving the Agent or appending Session events; queue edits accept only non-empty text content. Prompt admission consumes opaque receipts from the injected [`fileUploads`](../../client/file-upload/README.md) Host service and resolves every same-Agent receipt before sending the complete ordered content list through `ctx.attachments`. Prompt retries whose `requestId` is already queued or logged return the original acceptance without inserting another message. Create and fork are the only operations that create a new Agent directly. The service applies one preset-aware resume policy and subagent ownership fence to its own methods and to the Typert Agent and Session lookups used by other Remote namespaces. Queue mutation has one narrow exception: a live child whose current projected identity is continuable and comes from its own non-seed suffix accepts the ordinary Edit, Remove, and QueueDock Steer actions across both inbox destinations. One-shot, missing, unknown, corrupt, seed-only, or cold children remain rejected without resume. The skill catalog uses a live Agent's effective working directory and scoped registry when present. Cold listing uses the projected current directory, falling back to the recorded origin and then the configured directory default. It retains the preset standing scope without activating an Agent. The authenticated delivery routes use `workspaceDesktop()` for the serving Host name and file-manager behavior. `openWorkspacePath({ path, action: "reveal" })` delegates file-manager navigation to the native adapter; omitting `action` opens the file-type association, including HTML and SVG. Both operations require the composed filesystem to map the requested Host path to the same canonical process path; unmapped remote paths are refused before a native command runs. `session.projections` normally reads an exact baseline, including current-format cold logs, and returns `kind: sequenced`. Only a read requiring migration returns `kind: migration-required` hints without `asOfSeq`; missing keys remain absent. It returns null for a missing Session. Fork callers may set `allowMigration: false` to receive `session/migration-required` instead of starting source migration; omission retains automatic migration. The Client exposes shared values and explicit-read state through `projectionsBySession`; domains select their own keys. Session-list summaries carry `agentAvailable`, updated through existing summary and status events independently of durable projections. Sequenced reads and live projection frames use the same sequence ordering; cached reads fill hints without replacing sequenced values. +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` 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, 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. + 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 cc5a4ce4b2..b3cb5dccee 100644 --- a/packages/api/session-controller/README.zh.md +++ b/packages/api/session-controller/README.zh.md @@ -33,6 +33,8 @@ Client journal 在发布 follow 快照、live entry 或历史页之前验证当 每个 endpoint 都声明自己的激活策略。列表只读取持久化 header 与 projection cache row,绝不调用逐 Session stat 或打开冷 Session body。当前格式 cache identity 可以提供全部列表 hint;生命周期匹配的 predecessor cache 只能提供版本兼容的 title,作为可能过时的展示事实,绝不能作为权威 fold seed。搜索、附件、历史页、日志跟随、skill 发现和工作区路径打开可以在不激活 Agent 的情况下检查 persistence;`canOpenWorkspacePath()` 无需指定 Session 即可报告原生打开能力。取消要求 live 状态;queue 变更、模型、重命名、prompt 和文件引用操作可以解析或恢复普通 Session。提示词会在解析 Agent 或追加 Session 事件前,拒绝既没有非空白文本也没有附件的 content;queue edit 只接受非空文本 content。prompt 准入从注入的 [`fileUploads`](../../client/file-upload/README.zh.md) Host 服务取得不透明凭证,在把完整有序内容列表交给 `ctx.attachments` 前解析每个属于同一 Agent 的凭证。`requestId` 已进入 queue 或日志时,prompt 重试直接返回原来的接受结果,不会重复插入消息。只有 create 与 fork 会直接创建新 Agent。该服务把同一套感知 preset 的恢复策略和 subagent ownership fence 同时用于自身方法,以及其他 Remote namespace 使用的 Typert Agent 与 Session lookup。Queue 变更只有一个狭窄例外:当前 projection identity 为 continuable 且来自自身非 seed suffix 的在线 child,可以在两个 inbox 目标上使用普通 Edit、Remove 与 QueueDock Steer action。One-shot、缺失、未知、损坏、仅含 seed identity 或冷 child 继续被拒绝,且不会恢复。skill 目录优先使用已有 live Agent 的有效工作目录与限定范围内的注册表。冷态列表使用投影中的当前目录,依次回退到记录的起始目录与配置的默认目录。它保留 preset 常驻 scope,不会激活 Agent。经过鉴权的文件交付路由通过 `workspaceDesktop()` 获取提供服务的 Host 名称和文件管理器行为。`openWorkspacePath({ path, action: "reveal" })` 将文件管理器导航委托给原生适配器;省略 `action` 时按文件类型关联打开,包括 HTML 和 SVG。两种操作都要求当前文件系统将请求的 Host 路径映射到同一个规范进程路径;无法映射的远端路径会在执行原生命令之前被拒绝。 `session.projections` 正常读取准确基线,包括当前格式的冷日志,返回 `kind: sequenced`。只有需要迁移时才返回不带 `asOfSeq` 的 `kind: migration-required` 提示,缺少的键保持缺席。Session 不存在时返回 null。Fork 调用者可设置 `allowMigration: false`,在源历史需要迁移时收到 `session/migration-required`;省略则保持自动迁移。Client 通过 `projectionsBySession` 暴露共享值和显式读取状态,由领域选择自身的 key。Session 列表摘要携带 `agentAvailable`,通过已有摘要与状态事件更新,与持久化 projection 相互独立。有序读取与实时 projection 帧使用相同的序号排序规则;缓存读取只补充提示,不替换有序值。 +每个 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/control` 不读取资源,其调用访问 `global` 资源,需要 `global` role;对非 owner 调用方,`session/list` 只返回其 Session 被 `ctx.remoteAccess.allows()` 按同一 action 允许的行,`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。 + 列表行在状态可用时携带 `formatStatus`,Client 无需请求其投影即可识别需要迁移的 Session。 Host 仅在配置的工作时间片耗尽时,于完整条目之间让出事件循环;剩余工作较少时直接返回,不强制让出。每行及每次让出之后检查请求取消,取消时拒绝返回部分结果。每行同步计算;列表不承诺跨 Session 的统一快照。已经进入冷态队列的 Session 若在让出期间接入,仍返回缓存行:Client 对进行中请求的变更重放会纠正其可用性,带序号的投影会覆盖缓存提示。 diff --git a/packages/api/session-controller/package.json b/packages/api/session-controller/package.json index 436b151bbb..1bcdd19004 100644 --- a/packages/api/session-controller/package.json +++ b/packages/api/session-controller/package.json @@ -122,6 +122,7 @@ } }, "devDependencies": { + "@deepseek-ai/dsh-api-remote-access": "workspace:*", "@deepseek-ai/cordis": "workspace:~", "@deepseek-ai/dsh-agent": "workspace:*", "@deepseek-ai/dsh-agent-default-model": "workspace:*", diff --git a/packages/api/session-controller/src/index.ts b/packages/api/session-controller/src/index.ts index d3d5234e34..81fca45b9b 100644 --- a/packages/api/session-controller/src/index.ts +++ b/packages/api/session-controller/src/index.ts @@ -15,6 +15,8 @@ import type {} from '@deepseek-ai/dsh-session-projection-cache' import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query' import { Remote, RemoteError, TypertRemoteService, principalOf } from '@deepseek-ai/dsh-typert-protocol' import { canonicalClientTimeZone } from '@deepseek-ai/dsh-util-time' +import { assertNever } from '@deepseek-ai/dsh-util-values' +import type {} from '@deepseek-ai/dsh-api-remote-access' import { ApiSessionAgentController, inspectApiSession, @@ -33,6 +35,7 @@ import { ArchivedSessionGate } from './archived-session-gate.ts' import type { ModelCatalog, SessionWorkspacePathApplication, + SessionAddress, SessionAttachmentRequest, SessionAttachmentValue, SessionCancelRequest, @@ -77,6 +80,16 @@ declare module '@deepseek-ai/cordis' { } } +/** + * Read the Session resource a history address names: the Session itself, or + * the child Session whose transcript a subagent address reads. + * @param request - history request carrying the durable address. + * @returns the Session whose access the read requires. + */ +function addressedSession(request: { readonly address: SessionAddress }): SessionId { + return request.address.kind === 'session' ? request.address.sessionId : request.address.childSessionId +} + /** Session Controller deployment policy. */ export interface Config { /** Override platform desktop-opener detection. */ @@ -111,6 +124,7 @@ export class SessionController extends TypertRemoteService { 'fileUploads', 'fs', 'llm', + 'remoteAccess', 'sessions', 'sessionProjections', 'sessionQuery', @@ -142,6 +156,7 @@ export class SessionController extends TypertRemoteService { */ constructor(ctx: Context, config: Config, internals: SessionControllerInternals = {}) { super(ctx, 'sessionController', { namespace: 'session' }) + ctx.typert.resources.register('session', { field: 'sessionId' }) const resolved = SessionController.Config(config) installModelSelectionProjection(ctx) this.agents = new ApiSessionAgentController(ctx) @@ -259,7 +274,8 @@ export class SessionController extends TypertRemoteService { */ @Remote('list') async list(_request: SessionListRequest, signal: AbortSignal): Promise { - return { items: await this.listState.list(signal) } + const items = await this.listState.list(signal) + return { items: items.filter(item => this.ctx.remoteAccess.allows({ kind: 'session', id: item.sessionId })) } } /** @@ -502,7 +518,7 @@ export class SessionController extends TypertRemoteService { * @param signal - cancellation for persistence reads. * @returns one chronological page. */ - @Remote('page') + @Remote({ name: 'page', resources: { session: addressedSession } }) page(request: SessionPageRequest, signal: AbortSignal): Promise { return this.history.page(request, signal) } @@ -514,7 +530,7 @@ export class SessionController extends TypertRemoteService { * @returns a complete opening snapshot followed by gap-free durable event * frames and optional cursorless assistant-stream frames. */ - @Remote({ mode: 'stream' }) + @Remote({ mode: 'stream', resources: { session: addressedSession } }) follow(request: SessionFollowRequest, signal: AbortSignal): AsyncIterable { return this.history.follow(request, signal) } @@ -577,8 +593,23 @@ export class SessionController extends TypertRemoteService { * @returns one complete baseline followed by live replacement frames. */ @Remote({ mode: 'stream' }) - control(signal: AbortSignal): AsyncIterable { - return this.controlState.control(signal) + async *control(signal: AbortSignal): AsyncIterable { + for await (const frame of this.controlState.control(signal)) { + switch (frame.type) { + case 'baseline': { + const projections = Object.fromEntries(Object.entries(frame.value.projections) + .filter(([sessionId]) => this.ctx.remoteAccess.allows({ kind: 'session', id: sessionId as SessionId }))) + yield { type: 'baseline', value: { ...frame.value, projections } } + break + } + case 'projection': + if (this.ctx.remoteAccess.allows({ kind: 'session', id: frame.sessionId })) yield frame + break + /* v8 ignore next -- closed control-frame union. */ + default: + assertNever(frame) + } + } } diff --git a/packages/api/session-controller/tests/list-scheduling.host.spec.ts b/packages/api/session-controller/tests/list-scheduling.host.spec.ts index eefea76713..b3e412ec9f 100644 --- a/packages/api/session-controller/tests/list-scheduling.host.spec.ts +++ b/packages/api/session-controller/tests/list-scheduling.host.spec.ts @@ -8,6 +8,7 @@ import type { SessionRecord } from '@deepseek-ai/dsh-session-query' import { resolve } from 'node:path' import { performance } from 'node:perf_hooks' import { scheduler } from 'node:timers/promises' +import RemoteAccessService from '@deepseek-ai/dsh-api-remote-access' import SessionController from '../src/index.ts' import { afterEach, describe, expect, it, vi } from 'vitest' import { ApiSessionList } from '../src/list.ts' @@ -181,10 +182,15 @@ describe('Session-list work slices', () => { it('routes an overridden budget through the public Session Controller', async () => { const h = await harness([{ id: 'first', phase: 'cold' }, { id: 'middle', phase: 'cold' }, { id: 'last', phase: 'cold' }], 4) const dispose = (): void => {} - h.ctx.provide('typert', { lookups: { configure: () => dispose }, contexts: { configureHost: () => dispose } } as never) + h.ctx.provide('typert', { + lookups: { configure: () => dispose }, + contexts: { configureHost: () => dispose }, + resources: { register: () => dispose }, + } as never) h.ctx.provide('fileUploads', { registerAgentResolver: () => dispose } as never) h.ctx.provide('agentDefaultModel', { currentSelection: () => ({ provider: 'test', model: 'test' }) } as never) h.ctx.provide('workspaceRegistry', { list: () => [], archivedSessionIds: [] } as never) + new RemoteAccessService(h.ctx) const controller = new SessionController(h.ctx, { nativeOpen: false, listWorkSliceMs: 4 }, { canOpenPath: () => false }) const yieldWork = vi.spyOn(scheduler, 'yield') const result = await controller.list({}, new AbortController().signal) diff --git a/packages/api/session-controller/tests/remote-access.host.spec.ts b/packages/api/session-controller/tests/remote-access.host.spec.ts new file mode 100644 index 0000000000..04fa3b860f --- /dev/null +++ b/packages/api/session-controller/tests/remote-access.host.spec.ts @@ -0,0 +1,197 @@ +/** + * Session Controller under non-owner callers: the Session resource kind it + * registers, the Session readers of `session/page` and `session/follow`, + * `session/list` and `session/control` called through a `global` role and + * filtered per Session through `remoteAccess.allows()`, the caller recorded + * on prompt sources, and `session/projections` admitted per Session through + * the Gateway. + */ + +import { Context } from '@deepseek-ai/cordis' +import type { Agent } from '@deepseek-ai/dsh-agent' +import { mountAgentLoopTestDependencies, mountAgentLoopTestHarness } from '@deepseek-ai/dsh-agent-loop-testkit' +import TypertGatewayService from '@deepseek-ai/dsh-api-gateway' +import type { RemoteAccessBinding, RemoteAccessRole } from '@deepseek-ai/dsh-api-remote-access' +import { createUserMessage } from '@deepseek-ai/dsh-llm' +import { SessionId } from '@deepseek-ai/dsh-session' +import { remoteMethods, type PeerId, type PeerScope, type Principal, type RemoteInvocation } from '@deepseek-ai/dsh-typert-protocol' +import TypertRegistry from '@deepseek-ai/dsh-typert-registry' +import { afterEach, describe, expect, it } from 'vitest' +import type SessionController from '../src/index.ts' +import type { SessionControlFrame, SessionRequestId } from '../src/types.ts' +import { createSessionTestController } from './test-remote.ts' + +declare module '@deepseek-ai/dsh-typert-protocol' { + interface PrincipalMap { + /** Test-only non-owner caller shared by the access specs. */ + 'test-guest': { readonly kind: 'test-guest'; readonly guest: string } + } +} + +const OWNER: Principal = { kind: 'owner' } +const GUEST: Principal = { kind: 'test-guest', guest: 'session-controller' } +const SHARED = SessionId('shared-session') +const PRIVATE = SessionId('private-session') + +/** Calls the Session list and control methods; each item still needs a Session role. */ +const LISTER: RemoteAccessRole<'global'> = { name: 'test:session-lister', resource: 'global', actions: ['session/list', 'session/control'] } +/** Reads one Session: its list row, its control projections, and its stored projections. */ +const READER: RemoteAccessRole<'session'> = { + name: 'test:session-reader', resource: 'session', actions: ['session/list', 'session/control', 'session/projections'], +} +const LIST_CALLS: RemoteAccessBinding = { role: LISTER, resource: { kind: 'global' } } +const READS_SHARED: RemoteAccessBinding = { role: READER, resource: { kind: 'session', id: SHARED } } + +const owned = new Set() +afterEach(async () => { + await Promise.all([...owned].map(ctx => ctx.fiber.dispose())) + owned.clear() +}) + +/** + * Register the guest policy. + * @returns the bindings the policy answers with; edits apply from the next Remote call. + */ +function guestPolicy(ctx: Context, ...bindings: RemoteAccessBinding[]): RemoteAccessBinding[] { + const held = [...bindings] + ctx.remoteAccess.registerPolicy('test-guest', { bindings: () => held }) + return held +} + +/** Invoke `session/` as the guest through the Gateway. */ +function guestCall(ctx: Context, method: string, args: Readonly>): Promise { + const peer: PeerScope = { id: 'session-access-peer' as PeerId, ctx, principal: GUEST, signal: new AbortController().signal, dispose: async () => {} } + return ctx.typertGateway.invoke({ peer, namespace: 'session', method, args }) +} + +async function harness(): Promise<{ + readonly ctx: Context + readonly controller: SessionController + readonly shared: Agent + readonly hidden: Agent + /** The Context a Gateway dispatch of `session/` binds for `principal`. */ + readonly as: (principal: Principal, method: string) => Context +}> { + const ctx = new Context() + owned.add(ctx) + await ctx.plugin(TypertRegistry) + await mountAgentLoopTestDependencies(ctx) + const loop = await mountAgentLoopTestHarness(ctx) + const shared = await loop.create(SHARED) + const hidden = await loop.create(PRIVATE) + const controller = createSessionTestController(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' }) + const as = (principal: Principal, method: string): Context => { + const invocation: RemoteInvocation = { + request: { namespace: 'session', method, args: {} }, + service: 'sessionController', + peer: { id: 'session-access-peer' as PeerId, ctx, principal, signal: new AbortController().signal, dispose: async () => {} }, + signal: new AbortController().signal, + uplink: (): AsyncIterable => ({ [Symbol.asyncIterator]: () => ({ next: async () => ({ done: true, value: undefined }) }) }), + } + return ctx.extend({ invocation }) + } + return { ctx, controller, shared, hidden, as } +} + +function message(text: string) { + return createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'user' } }) +} + +describe('Session Controller access', () => { + it('owns the Session resource kind and its default sessionId field', async () => { + const { ctx } = await harness() + expect(() => ctx.typert.resources.register('session', { field: 'sessionId' })).toThrow(/already registered/) + }) + + it('reads the Session resource of page and follow requests from the addressed Session', async () => { + const { controller } = await harness() + for (const method of ['page', 'follow']) { + const reader = remoteMethods(controller).find(marker => marker.method === method)?.resources?.['session'] + expect(reader?.({ address: { kind: 'session', sessionId: SHARED } })).toBe(SHARED) + // A subagent address reads the child transcript; whether the parent's access covers it is the policy's rule. + expect(reader?.({ address: { kind: 'subagent', parentSessionId: SHARED, childSessionId: PRIVATE } })).toBe(PRIVATE) + } + }) + + it('lets a global role call session/list and lists only the Sessions a Session role admits', async () => { + const { ctx, as } = await harness() + await ctx.plugin(TypertGatewayService) + const held = guestPolicy(ctx, LIST_CALLS, READS_SHARED) + const rows = async (): Promise => { + const value = await guestCall(ctx, 'list', { _request: {} }) as { readonly items: readonly { readonly sessionId: SessionId }[] } + return value.items.map(item => item.sessionId) + } + await expect(rows()).resolves.toEqual([SHARED]) + // The global role admits the call, not any row. + held.splice(0, held.length, LIST_CALLS) + await expect(rows()).resolves.toEqual([]) + // A Session role alone does not admit the call: the global resource does not inherit from it. + held.splice(0, held.length, READS_SHARED) + await expect(rows()).rejects.toMatchObject({ + code: 'remote-access/denied', details: { action: 'session/list', reason: 'no-allow' }, + }) + const owner = await as(OWNER, 'list').sessionController.list({}, new AbortController().signal) + expect(owner.items.map(item => item.sessionId).sort()).toEqual([PRIVATE, SHARED].sort()) + }) + + it('filters session/control baselines and projection frames for a non-owner', async () => { + const { ctx, shared, hidden, as } = await harness() + guestPolicy(ctx, LIST_CALLS, READS_SHARED) + const abort = new AbortController() + const guest = as(GUEST, 'control').sessionController.control(abort.signal)[Symbol.asyncIterator]() + const owner = as(OWNER, 'control').sessionController.control(abort.signal)[Symbol.asyncIterator]() + const baselineSessions = async (iterator: AsyncIterator): Promise => { + const next = await iterator.next() + if (next.done === true || next.value.type !== 'baseline') throw new Error('control stream opened without a baseline') + return Object.keys(next.value.value.projections).sort() + } + expect(await baselineSessions(guest)).toEqual([SHARED]) + expect(await baselineSessions(owner)).toEqual([PRIVATE, SHARED].sort()) + + hidden.inbox.append('next-turn', message('private work')) + shared.inbox.append('next-turn', message('shared work')) + const projected = async (iterator: AsyncIterator): Promise => { + for (;;) { + const next = await iterator.next() + if (next.done === true) throw new Error('control stream ended') + if (next.value.type === 'projection' && next.value.key === 'inbox') return next.value.sessionId + } + } + // The owner sees the private Session's frame first; the guest's first Inbox frame is the shared one. + expect(await projected(owner)).toBe(PRIVATE) + expect(await projected(guest)).toBe(SHARED) + abort.abort() + await Promise.all([guest.return?.(), owner.return?.()]) + }) + + it('records a non-owner caller on the prompt source and leaves owner sources unchanged', async () => { + const { shared, as } = await harness() + const signal = new AbortController().signal + const prompt = (principal: Principal, requestId: string) => as(principal, 'prompt').sessionController.prompt({ + sessionId: SHARED, + requestId: requestId as SessionRequestId, + mode: 'queue', + content: [{ type: 'text', text: `from ${principal.kind}` }], + }, signal) + await expect(prompt(GUEST, 'guest-1')).resolves.toEqual({ accepted: true }) + await expect(prompt(OWNER, 'owner-1')).resolves.toEqual({ accepted: true }) + // The durable Inbox splice records each accepted message with its source. + const logged = shared.session.snapshotEvents() + .flatMap(event => event.type === 'agent/inbox/spliced' ? event.data.inserted.map(item => item.source) : []) + expect(logged).toEqual([ + { kind: 'user', rpcId: 'guest-1', principal: GUEST }, + { kind: 'user', rpcId: 'owner-1' }, + ]) + }) + + it('lets a non-owner read projections only in a Session its Session role names', async () => { + const { ctx } = await harness() + await ctx.plugin(TypertGatewayService) + guestPolicy(ctx, READS_SHARED) + const projections = (sessionId: SessionId): Promise => guestCall(ctx, 'projections', { request: { sessionId } }) + await expect(projections(SHARED)).resolves.toMatchObject({ kind: 'sequenced' }) + await expect(projections(PRIVATE)).rejects.toMatchObject({ + code: 'remote-access/denied', details: { action: 'session/projections', reason: 'no-allow' }, + }) + }) +}) diff --git a/packages/api/session-controller/tests/test-remote.ts b/packages/api/session-controller/tests/test-remote.ts index 71db06b22b..586afcde7b 100644 --- a/packages/api/session-controller/tests/test-remote.ts +++ b/packages/api/session-controller/tests/test-remote.ts @@ -31,6 +31,7 @@ import { remoteErrorOf, type RemoteResult, } from '@deepseek-ai/dsh-typert-protocol' +import RemoteAccessService from '@deepseek-ai/dsh-api-remote-access' import SessionController from '../src/index.ts' import type { ModelCatalog, @@ -243,6 +244,7 @@ function installControllers( ctx.provide('typert', { lookups: { configure: () => dispose }, contexts: { configureHost: () => dispose }, + resources: { register: () => dispose }, } as never) } if (ctx.get('agentDefaultModel') === undefined) { @@ -289,6 +291,7 @@ function installControllers( } as never) } installSessionReadTestServices(ctx) + if (ctx.get('remoteAccess') === undefined) new RemoteAccessService(ctx) const cwd = vi.spyOn(process, 'cwd').mockReturnValue(defaults.cwd) let controller: SessionController try { diff --git a/packages/api/session-controller/tsconfig.host.json b/packages/api/session-controller/tsconfig.host.json index 6376b437ca..8f90ca7a6a 100644 --- a/packages/api/session-controller/tsconfig.host.json +++ b/packages/api/session-controller/tsconfig.host.json @@ -104,6 +104,9 @@ { "path": "../../typert/protocol" }, + { + "path": "../remote-access/tsconfig.host.json" + }, { "path": "../../typert/registry" }, diff --git a/packages/api/workspace-controller/README.i18n.yaml b/packages/api/workspace-controller/README.i18n.yaml index d804399b83..e3cb08a719 100644 --- a/packages/api/workspace-controller/README.i18n.yaml +++ b/packages/api/workspace-controller/README.i18n.yaml @@ -15,8 +15,8 @@ en: 4e53164159b5ce41 zh: 89582dc0294d0776 /workspace-controller/use-this-package: - en: aa0d93826ab685fe - zh: 144e48d49020f5a4 + en: 8276c5edb7d0db7d + zh: cda0c91f8c3776e6 /workspace-controller/use-this-package/first-use-workspace: en: 61032e060e7213f0 zh: 437eb982a8738539 diff --git a/packages/api/workspace-controller/README.md b/packages/api/workspace-controller/README.md index b24053384f..27c26c7d9b 100644 --- a/packages/api/workspace-controller/README.md +++ b/packages/api/workspace-controller/README.md @@ -24,6 +24,8 @@ English | [中文](README.zh.md) The Host controller serializes mutations whose correctness depends on current registry state and throws `RemoteError` with a stable error code for expected failures. Its `follow()` stream synchronously attaches to durable Workspace changes, emits one complete baseline first, then emits ordered `upsert`, `remove`, `order`, `archived`, and `pinned` increments. Archive and pin sets are Session id arrays, with the most recently pinned id first in the pin array. A reconnect starts another generation with a replacement baseline, so consumers do not depend on receiving every increment while disconnected. `archiveSession` without `stopActivity` refuses a Session with running work as `workspace/session-active`, whose details list that work by family (`turn`, `subagent`, `job`, `schedule`) with item ids and labels; with `stopActivity: true` the registry's providers stop the work first and the response arrives once the archive set is durable, while the stops settle in the background. +The controller injects `remoteAccess` and registers the `workspace` resource kind with the default field `workspaceId` through `typert.resources`. `workspace/follow` reads no resource, so its call touches the `global` resource: a Peer whose principal is not the owner may follow it only when its principal kind's policy in [`dsh-api-remote-access`](../remote-access/README.md) binds a `global` role listing the action. For that caller, the baseline, `upsert` frames, and `order` ids keep only the Workspaces `ctx.remoteAccess.allows()` admits as `workspace` resources, and only the Session ids it admits as `session` resources, within each Workspace and in the archived and pinned sets, so `workspace` and `session` roles list the same action. The stream tracks the Workspaces it delivered. When one is no longer admitted at its next change or in a new order, the stream sends a `replace` increment carrying the complete filtered state instead of a `remove`, because a removed id is never resurrected on the Client; `replace` is an ordinary increment that the Client applies like an opening baseline, not the start of a new generation, so it neither reports an error nor reconnects; once access returns, the Workspace reappears through an ordinary `upsert`. A `remove` frame reports a deletion and reaches only a caller that received that Workspace. Owner frames are unfiltered. + The Client entry provides `ClientWorkspaceModel` and `createWorkspaceStateStream()`. The model owns Workspace rows, registry order, archived and pinned Session identities, unary mutation echoes, and stream/unary race resolution. A newer Host row wins by `updatedAt`; a committed stream order outranks an older unary response; a removed Workspace id cannot be resurrected by delayed data. Pin snapshots change only when their identities or order change. The package exposes framework-neutral snapshots and subscriptions, leaving navigation policy and React hooks to the UI owner. `WorkspaceController.archiveSession(sessionId, { stopActivity })` throws `WorkspaceArchiveError` with the Host's `rpcError`, so a surface can tell the running-work refusal from a missing Session or a carrier fault and offer to stop the work. diff --git a/packages/api/workspace-controller/README.zh.md b/packages/api/workspace-controller/README.zh.md index dc240950cc..4772d0c9b1 100644 --- a/packages/api/workspace-controller/README.zh.md +++ b/packages/api/workspace-controller/README.zh.md @@ -24,6 +24,8 @@ kind: "package-reference" Host 控制器会串行执行正确性取决于当前注册表状态的变更,并为预期失败抛出带有稳定错误码的 `RemoteError`。它的 `follow()` 流会同步订阅持久 Workspace 变更,先发出一份完整 baseline,再按顺序发出 `upsert`、`remove`、`order`、`archived` 和 `pinned` 增量。归档与置顶集合都是会话 id 数组,置顶数组把最近置顶的 id 放在前面。重连会以替换 baseline 开始新一代,因此消费方不依赖收到断线期间的每个增量。不带 `stopActivity` 的 `archiveSession` 会以 `workspace/session-active` 拒绝仍有工作在跑的会话,其 details 按族(`turn`、`subagent`、`job`、`schedule`)列出这些工作及各项的 id 与名称;带 `stopActivity: true` 时注册表的提供方先停止这些工作,归档集合持久化后即返回响应,停止在后台收敛。 +控制器注入 `remoteAccess`,并经 `typert.resources` 以默认字段 `workspaceId` 登记 `workspace` 资源种类。`workspace/follow` 不读取资源,其调用访问 `global` 资源:principal 不是 owner 的 Peer 只有在其 principal kind 于 [`dsh-api-remote-access`](../remote-access/README.zh.md) 中的 policy 绑定了列出该 action 的 `global` role 时才能订阅它。对这类调用方,baseline、`upsert` 帧与 `order` 中的 id 只保留 `ctx.remoteAccess.allows()` 作为 `workspace` 资源允许的 Workspace,以及它作为 `session` 资源允许的会话 id(各 Workspace 内的会话以及归档与置顶集合),因此 `workspace` role 与 `session` role 也列出同一 action。流记录已下发的 Workspace。某个 Workspace 在下一次变化或新顺序中已不被允许时,流发送一个携带完整过滤后状态的 `replace` 增量帧而不是 `remove`,因为 Client 不会复活已移除的 id;`replace` 是普通增量帧,Client 像开场 baseline 一样整体替换状态,但它不开启新的一代,因此既不报告错误也不重连;访问恢复后,该 Workspace 经普通 `upsert` 重新出现。`remove` 帧表示删除,只发给收到过该 Workspace 的调用方。owner 的帧不过滤。 + Client 入口提供 `ClientWorkspaceModel` 和 `createWorkspaceStateStream()`。该模型拥有 Workspace 行、registry 顺序、归档与置顶会话身份、一元变更回显,以及流与一元调用的竞态处理。较新的 Host 行按 `updatedAt` 获胜;已提交的流顺序优先于较旧的一元响应;已经移除的 Workspace id 不会被延迟数据复活。置顶快照仅在会话身份或顺序变化时更新。该包公开与框架无关的快照和订阅,把导航策略与 React 钩子留给 UI owner。`WorkspaceController.archiveSession(sessionId, { stopActivity })` 抛出携带 Host `rpcError` 的 `WorkspaceArchiveError`,界面因此能区分"仍有工作在跑"的拒绝与会话缺失或载体故障,并提议停止这些工作。 diff --git a/packages/api/workspace-controller/package.json b/packages/api/workspace-controller/package.json index ac9d45a3e1..a6105a5c1c 100644 --- a/packages/api/workspace-controller/package.json +++ b/packages/api/workspace-controller/package.json @@ -89,6 +89,7 @@ "devDependencies": { "@deepseek-ai/cordis": "workspace:~", "@deepseek-ai/dsh-api-gateway": "workspace:*", + "@deepseek-ai/dsh-api-remote-access": "workspace:*", "@deepseek-ai/dsh-client-connection": "workspace:*", "@deepseek-ai/dsh-client-store": "workspace:*", "@deepseek-ai/dsh-client-test-runtime": "workspace:*", diff --git a/packages/api/workspace-controller/src/client/index.ts b/packages/api/workspace-controller/src/client/index.ts index c3182102a7..3bbf4670a0 100644 --- a/packages/api/workspace-controller/src/client/index.ts +++ b/packages/api/workspace-controller/src/client/index.ts @@ -98,6 +98,9 @@ export function createWorkspaceStateStream( function acceptIncrement(accept: WorkspaceFollowSink, frame: WorkspaceFollowIncrement): void { switch (frame.type) { + case 'replace': + accept.replaceBaseline(frame.value) + return case 'upsert': accept.upsertView(frame.workspace) return diff --git a/packages/api/workspace-controller/src/feed.ts b/packages/api/workspace-controller/src/feed.ts index 709f17fa5d..be1cbdb21c 100644 --- a/packages/api/workspace-controller/src/feed.ts +++ b/packages/api/workspace-controller/src/feed.ts @@ -15,6 +15,9 @@ import type { WorkspaceView, } from './types.ts' +/** The frames the feed produces: every follow frame except `replace`, which only access filtering sends. */ +type WorkspaceFeedFrame = Exclude + /** * Project one authoritative Workspace entity into its Remote value. * @param workspace - authoritative registry entity. @@ -82,7 +85,7 @@ export class WorkspaceFeed { * @param signal - generation cancellation. * @returns baseline followed by ordered Workspace increments. */ - async *follow(signal: AbortSignal): AsyncIterable { + async *follow(signal: AbortSignal): AsyncIterable { signal.throwIfAborted() const follower = new WorkspaceFollower() this.followers.add(follower) @@ -138,7 +141,7 @@ export class WorkspaceFeed { }) } - private publish(frame: Exclude): void { + private publish(frame: Exclude): void { for (const follower of this.followers) follower.push(frame) } } @@ -148,11 +151,11 @@ function sameStrings(left: readonly string[], right: readonly string[]): boolean } class WorkspaceFollower { - private readonly frames = new Deque() + private readonly frames = new Deque() private waiting: (() => void) | undefined private closed = false - push(frame: WorkspaceFollowFrame): void { + push(frame: WorkspaceFeedFrame): void { /* v8 ignore next -- closed followers are removed before later publication can reach them. */ if (this.closed) return this.frames.pushBack(frame) @@ -165,7 +168,7 @@ class WorkspaceFollower { this.waiting?.() } - async *read(signal: AbortSignal): AsyncIterable { + async *read(signal: AbortSignal): AsyncIterable { while (!this.closed && !signal.aborted) { const frame = this.frames.popFront() if (frame !== undefined) { diff --git a/packages/api/workspace-controller/src/index.ts b/packages/api/workspace-controller/src/index.ts index 0b23517921..f275eb3d12 100644 --- a/packages/api/workspace-controller/src/index.ts +++ b/packages/api/workspace-controller/src/index.ts @@ -3,6 +3,8 @@ import { Context } from '@deepseek-ai/cordis' import z from '@deepseek-ai/schemastery' import { Remote, TypertRemoteService } from '@deepseek-ai/dsh-typert-protocol' +import type {} from '@deepseek-ai/dsh-api-remote-access' +import type { SessionId } from '@deepseek-ai/dsh-session/types' import { WorkspaceCommands } from './commands.ts' import { DirectoryPickerController } from './directory-picker.ts' import { WorkspaceFeed, workspaceView } from './feed.ts' @@ -10,11 +12,13 @@ import { defaultWorkspaceDirectory, validateDocumentsDirectory } from './default import type { WorkspaceArchiveSessionRequest, WorkspaceArchiveValue, + WorkspaceBaseline, WorkspaceCreateRequest, WorkspaceCreateValue, WorkspaceDeleteRequest, WorkspaceDeleteValue, WorkspaceFollowFrame, + WorkspaceId, WorkspaceInsertBeforeRequest, WorkspaceInsertSessionBeforeRequest, WorkspaceOrderValue, @@ -49,7 +53,7 @@ declare module '@deepseek-ai/cordis' { /** Host service backing the generated `ctx.remote.workspace` namespace. */ export class WorkspaceController extends TypertRemoteService { - static inject = ['typert', 'workspaceRegistry'] + static inject = ['remoteAccess', 'typert', 'workspaceRegistry'] static Config: z = z.object({ documentsDirectory: z.string(), @@ -66,6 +70,7 @@ export class WorkspaceController extends TypertRemoteService { */ constructor(ctx: Context, config: Config = {}) { super(ctx, 'workspaceController', { namespace: 'workspace' }) + ctx.typert.resources.register('workspace', { field: 'workspaceId' }) this.config = WorkspaceController.Config(config) if (this.config.documentsDirectory !== undefined) validateDocumentsDirectory(this.config.documentsDirectory) this.commands = new WorkspaceCommands(ctx) @@ -187,14 +192,77 @@ export class WorkspaceController extends TypertRemoteService { } /** - * Stream a complete Workspace baseline followed by ordered increments. + * Stream a complete Workspace baseline followed by ordered increments. A + * non-owner receives only the Workspaces and Sessions `allows()` admits. + * When a delivered Workspace is no longer admitted at its next change or + * the next order change, the stream sends a `replace` frame with the + * complete filtered state instead of a `remove`, because a Client never + * resurrects a removed Workspace; a `remove` frame reports a deletion and + * reaches only a Client that received that Workspace. * @param signal - generation cancellation. * @returns baseline followed by ordered Workspace increments. */ @Remote({ mode: 'stream' }) - follow(signal: AbortSignal): AsyncIterable { - return this.feed.follow(signal) + async *follow(signal: AbortSignal): AsyncIterable { + const access = this.ctx.remoteAccess + const visible = (workspaceId: WorkspaceId): boolean => access.allows({ kind: 'workspace', id: workspaceId }) + const sessions = (sessionIds: readonly SessionId[]): SessionId[] => + sessionIds.filter(sessionId => access.allows({ kind: 'session', id: sessionId })) + /** Workspaces this generation has delivered and not yet removed. */ + const delivered = new Set() + /** The admitted part of `value`; it becomes the delivered set. */ + const filtered = (value: WorkspaceBaseline): WorkspaceBaseline => { + const items = value.items.filter(item => visible(item.workspaceId)) + delivered.clear() + for (const item of items) delivered.add(item.workspaceId) + return { + items: items.map(item => ({ ...item, sessionIds: sessions(item.sessionIds) })), + archivedSessionIds: sessions(value.archivedSessionIds), + pinnedSessionIds: sessions(value.pinnedSessionIds), + } + } + for await (const frame of this.feed.follow(signal)) { + switch (frame.type) { + case 'baseline': + yield { type: 'baseline', value: filtered(frame.value) } + break + case 'upsert': { + const workspaceId = frame.workspace.workspaceId + if (visible(workspaceId)) { + delivered.add(workspaceId) + yield { type: 'upsert', workspace: { ...frame.workspace, sessionIds: sessions(frame.workspace.sessionIds) } } + } else if (delivered.has(workspaceId)) { + yield { type: 'replace', value: filtered(this.feed.baseline()) } + } + break + } + case 'remove': + if (delivered.delete(frame.workspaceId)) yield frame + break + case 'order': + if (frame.workspaceIds.some(workspaceId => delivered.has(workspaceId) && !visible(workspaceId))) { + yield { type: 'replace', value: filtered(this.feed.baseline()) } + } else { + yield { type: 'order', workspaceIds: frame.workspaceIds.filter(visible) } + } + break + case 'archived': + yield { type: 'archived', archivedSessionIds: sessions(frame.archivedSessionIds) } + break + case 'pinned': + yield { type: 'pinned', pinnedSessionIds: sessions(frame.pinnedSessionIds) } + break + /* v8 ignore next -- closed follow-frame union. */ + default: + assertNever(frame) + } + } } } +/* v8 ignore next 3 -- closed-union backstop. */ +function assertNever(value: never): never { + throw new Error(`unreachable Workspace follow frame: ${JSON.stringify(value)}`) +} + export default WorkspaceController diff --git a/packages/api/workspace-controller/src/types.ts b/packages/api/workspace-controller/src/types.ts index d5b98746be..cdf9adf073 100644 --- a/packages/api/workspace-controller/src/types.ts +++ b/packages/api/workspace-controller/src/types.ts @@ -157,8 +157,13 @@ export interface WorkspaceBaseline { readonly pinnedSessionIds: readonly SessionId[] } -/** One ordered Workspace change after a generation's baseline. */ +/** + * One ordered Workspace change after a generation's baseline. `replace` + * carries the complete state that supersedes everything delivered before it + * in the generation; it is a change, not a new generation's opening frame. + */ export type WorkspaceFollowIncrement = + | { readonly type: 'replace'; readonly value: WorkspaceBaseline } | { readonly type: 'upsert'; readonly workspace: WorkspaceView } | { readonly type: 'remove'; readonly workspaceId: WorkspaceId } | { readonly type: 'order'; readonly workspaceIds: readonly WorkspaceId[] } diff --git a/packages/api/workspace-controller/tests/transport.client.spec.ts b/packages/api/workspace-controller/tests/transport.client.spec.ts index b74ddfb394..b1d6ca2a9b 100644 --- a/packages/api/workspace-controller/tests/transport.client.spec.ts +++ b/packages/api/workspace-controller/tests/transport.client.spec.ts @@ -183,6 +183,38 @@ describe('Workspace state stream', () => { expect(streamStates(mock)).toEqual(['cancelled']) }) + it('applies a replace increment in place, so a Workspace hidden and shown again never errors or reconnects', async ({ mock, start }) => { + const { remote } = await gatewayClient(mock, start) + mock.stream(FOLLOW, openStream([baseline('one', 'two')])) + const model = new ClientWorkspaceModel(remote.workspace) + const carrierFailed = vi.fn(() => { model.handleCarrierFailure() }) + const failed = vi.fn((error: unknown) => { model.handleStreamFailure(error) }) + const stream = createWorkspaceStateStream(remote, { accept: model, carrierFailed, failed }) + const states = new Set() + model.subscribe(() => { states.add(model.getSnapshot().state) }) + const ids = (): string[] => model.getSnapshot().items.map(item => item.workspaceId).sort() + const replace = (...workspaceIds: string[]): WorkspaceFollowFrame => ({ type: 'replace', value: baseline(...workspaceIds).value }) + + stream.start() + await vi.waitFor(() => { expect(ids()).toEqual(['one', 'two']) }) + // Access to `two` is revoked, restored through a replace, revoked again, and restored through an upsert. + mock.streams.push(FOLLOW, replace('one')) + await vi.waitFor(() => { expect(ids()).toEqual(['one']) }) + mock.streams.push(FOLLOW, replace('one', 'two')) + await vi.waitFor(() => { expect(ids()).toEqual(['one', 'two']) }) + mock.streams.push(FOLLOW, replace('one')) + await vi.waitFor(() => { expect(ids()).toEqual(['one']) }) + mock.streams.push(FOLLOW, { type: 'upsert', workspace: workspace('two') }) + await vi.waitFor(() => { expect(ids()).toEqual(['one', 'two']) }) + + expect(states).not.toContain('error') + expect(model.getSnapshot()).toMatchObject({ phase: 'ready', state: 'idle', error: null }) + expect(carrierFailed).not.toHaveBeenCalled() + expect(failed).not.toHaveBeenCalled() + expect(streamStates(mock)).toEqual(['open']) + await stream.dispose() + }) + it('retains the old state across carrier loss and applies the replacement baseline', async ({ mock, start }) => { const { remote } = await gatewayClient(mock, start) const carrier = new RemoteStreamCarrierError('socket lost') diff --git a/packages/api/workspace-controller/tests/workspace-controller.host.spec.ts b/packages/api/workspace-controller/tests/workspace-controller.host.spec.ts index 35158490f2..f284f35155 100644 --- a/packages/api/workspace-controller/tests/workspace-controller.host.spec.ts +++ b/packages/api/workspace-controller/tests/workspace-controller.host.spec.ts @@ -6,9 +6,10 @@ import { Context } from '@deepseek-ai/cordis' import SessionStore, { SessionId } from '@deepseek-ai/dsh-session' import Storage from '@deepseek-ai/dsh-storage' import { DomainFacility } from '@deepseek-ai/dsh-storage-domain' -import { RemoteError } from '@deepseek-ai/dsh-typert-protocol' +import { RemoteError, type PeerId, type Principal, type RemoteInvocation } from '@deepseek-ai/dsh-typert-protocol' import WorkspaceRegistry from '@deepseek-ai/dsh-workspace' import type { WorkspaceId } from '@deepseek-ai/dsh-workspace/types' +import RemoteAccessService, { type RemoteAccessCondition, type RemoteAccessRole } from '@deepseek-ai/dsh-api-remote-access' import WorkspaceController from '../src/index.ts' import { DEFAULT_WORKSPACE_DIRECTORY } from '../src/default-workspace.ts' import { WorkspaceFeed } from '../src/feed.ts' @@ -67,11 +68,39 @@ async function harness(options: { systemDocuments?: boolean } = {}) { ctx.provide('typert', { lookups: { configure: () => dispose }, contexts: { configureHost: () => dispose }, + resources: { register: vi.fn(() => dispose) }, } as never) + new RemoteAccessService(ctx) const controller = new WorkspaceController(ctx, options.systemDocuments === true ? {} : { documentsDirectory: root }) return { controller, ctx, root, storageDomain } } +declare module '@deepseek-ai/dsh-typert-protocol' { + interface PrincipalMap { + /** Test-only non-owner caller shared by the access specs. */ + 'test-guest': { readonly kind: 'test-guest'; readonly guest: string } + } +} + +/** Opens the Workspace follow stream; each Workspace and Session in it still needs its own role. */ +const FOLLOWER: RemoteAccessRole<'global'> = { name: 'test:workspace-follower', resource: 'global', actions: ['workspace/follow'] } +/** Shows one Workspace in the follow stream. */ +const WORKSPACE_VIEWER: RemoteAccessRole<'workspace'> = { name: 'test:workspace-viewer', resource: 'workspace', actions: ['workspace/follow'] } +/** Shows one Session id in the follow stream. */ +const SESSION_VIEWER: RemoteAccessRole<'session'> = { name: 'test:session-viewer', resource: 'session', actions: ['workspace/follow'] } + +/** The Context a Gateway dispatch of `workspace/follow` binds for `principal`. */ +function followerContext(ctx: Context, principal: Principal): Context { + const invocation: RemoteInvocation = { + request: { namespace: 'workspace', method: 'follow', args: {} }, + service: 'workspaceController', + peer: { id: 'workspace-access-peer' as PeerId, ctx, principal, signal: new AbortController().signal, dispose: async () => {} }, + signal: new AbortController().signal, + uplink: (): AsyncIterable => ({ [Symbol.asyncIterator]: () => ({ next: async () => ({ done: true, value: undefined }) }) }), + } + return ctx.extend({ invocation }) +} + function stageDir(root: string, name: string): string { const path = join(root, name) mkdirSync(path, { recursive: true }) @@ -416,6 +445,110 @@ describe('WorkspaceController follow', () => { await expect(iterator.next()).resolves.toEqual({ done: true, value: undefined }) }) + it('follows only the Workspaces and Sessions a non-owner may follow', async () => { + const { controller, ctx, root } = await harness() + // oxlint-disable-next-line typescript/unbound-method -- the assertion inspects the spy and never calls it. + expect(ctx.typert.resources.register).toHaveBeenCalledWith('workspace', { field: 'workspaceId' }) + const shown = await controller.create({ path: stageDir(root, 'shown') }) + const hiddenWorkspace = await controller.create({ path: stageDir(root, 'hidden') }) + const shownId = shown.workspace.workspaceId + const hiddenId = hiddenWorkspace.workspace.workspaceId + const shared = ctx.sessions.create(SessionId('shared'), { meta: { cwd: shown.workspace.path } }) + const secret = ctx.sessions.create(SessionId('secret'), { meta: { cwd: shown.workspace.path } }) + const workspace = ctx.workspaceRegistry.get(shownId) + if (workspace === undefined) throw new Error('fixture Workspace disappeared') + await workspace.attachSession(shared.id) + await workspace.attachSession(secret.id) + await controller.pinSession({ sessionId: shared.id }) + await controller.pinSession({ sessionId: secret.id }) + ctx.remoteAccess.registerPolicy('test-guest', { + bindings: () => [ + { role: FOLLOWER, resource: { kind: 'global' } }, + { role: WORKSPACE_VIEWER, resource: { kind: 'workspace', id: shownId } }, + { role: SESSION_VIEWER, resource: { kind: 'session', id: shared.id } }, + ], + }) + const guest: Principal = { kind: 'test-guest', guest: 'workspace-controller' } + const abort = new AbortController() + const iterator = followerContext(ctx, guest).workspaceController.follow(abort.signal)[Symbol.asyncIterator]() + await expect(nextFrame(iterator)).resolves.toEqual({ + type: 'baseline', + value: { + items: [expect.objectContaining({ workspaceId: shownId, sessionIds: [shared.id] })], + archivedSessionIds: [], + pinnedSessionIds: [shared.id], + }, + }) + + await controller.rename({ workspaceId: hiddenId, title: 'still hidden' }) + await controller.rename({ workspaceId: shownId, title: 'renamed' }) + await expect(nextFrame(iterator)).resolves.toMatchObject({ + type: 'upsert', workspace: { workspaceId: shownId, title: 'renamed', sessionIds: [shared.id] }, + }) + await controller.unpinSession({ sessionId: secret.id }) + await expect(nextFrame(iterator)).resolves.toEqual({ type: 'pinned', pinnedSessionIds: [shared.id] }) + await controller.unpinSession({ sessionId: shared.id }) + await controller.archiveSession({ sessionId: secret.id }) + await controller.archiveSession({ sessionId: shared.id }) + await expect(nextFrame(iterator)).resolves.toEqual({ type: 'pinned', pinnedSessionIds: [] }) + await expect(nextFrame(iterator)).resolves.toEqual({ type: 'archived', archivedSessionIds: [] }) + await expect(nextFrame(iterator)).resolves.toEqual({ type: 'archived', archivedSessionIds: [shared.id] }) + await controller.delete({ workspaceId: hiddenId }) + await expect(nextFrame(iterator)).resolves.toEqual({ type: 'order', workspaceIds: [shownId] }) + await controller.delete({ workspaceId: shownId }) + await expect(nextFrame(iterator)).resolves.toEqual({ type: 'order', workspaceIds: [] }) + await expect(nextFrame(iterator)).resolves.toEqual({ type: 'remove', workspaceId: shownId }) + + abort.abort() + await expect(iterator.next()).resolves.toEqual({ done: true, value: undefined }) + }) + + it('sends a non-owner a replace increment when a delivered Workspace loses visibility, so it can reappear', async () => { + const { controller, ctx, root } = await harness() + const first = (await controller.create({ path: stageDir(root, 'first') })).workspace.workspaceId + const second = (await controller.create({ path: stageDir(root, 'second') })).workspace.workspaceId + // One Remote call reads its bindings once, so visibility changes during the stream come from the condition. + const allowedWorkspaces = new Set([first, second]) + const stillShared: RemoteAccessCondition<'workspace'> = ({ resource }) => + allowedWorkspaces.has(resource.id) ? { kind: 'allow' } : { kind: 'deny', reason: 'test:unshared' } + ctx.remoteAccess.registerPolicy('test-guest', { + bindings: () => [ + { role: FOLLOWER, resource: { kind: 'global' } }, + ...[first, second].map(id => ({ role: WORKSPACE_VIEWER, resource: { kind: 'workspace', id } as const, condition: stillShared })), + ], + }) + const abort = new AbortController() + const iterator = followerContext(ctx, { kind: 'test-guest', guest: 'withdrawn' }).workspaceController.follow(abort.signal)[Symbol.asyncIterator]() + const ids = (frame: WorkspaceFollowFrame): string[] => + frame.type === 'baseline' || frame.type === 'replace' ? frame.value.items.map(item => item.workspaceId).sort() : [] + const opening = await nextFrame(iterator) + expect(opening.type).toBe('baseline') + expect(ids(opening)).toEqual([first, second].sort()) + + // Revoked: the next change replaces the state instead of sending a `remove` the Client would never undo. + allowedWorkspaces.delete(first) + await controller.rename({ workspaceId: first, title: 'revoked' }) + const withoutFirst = await nextFrame(iterator) + expect(withoutFirst.type).toBe('replace') + expect(ids(withoutFirst)).toEqual([second]) + allowedWorkspaces.delete(second) + const third = (await controller.create({ path: stageDir(root, 'third') })).workspace.workspaceId + const empty = await nextFrame(iterator) + expect(empty.type).toBe('replace') + expect(ids(empty)).toEqual([]) + // A deletion reaches only a Client that received the Workspace. + await controller.delete({ workspaceId: third }) + await expect(nextFrame(iterator)).resolves.toEqual({ type: 'order', workspaceIds: [] }) + + // Restored: the Workspace reappears through an ordinary upsert. + allowedWorkspaces.add(first) + await controller.rename({ workspaceId: first, title: 'restored' }) + await expect(nextFrame(iterator)).resolves.toMatchObject({ type: 'upsert', workspace: { workspaceId: first, title: 'restored' } }) + + abort.abort() + await expect(iterator.next()).resolves.toEqual({ done: true, value: undefined }) + }) + it('ignores unrelated domain writes and closes active followers on disposal', async () => { const { controller, ctx, root } = await harness() const abort = new AbortController() diff --git a/packages/api/workspace-controller/tsconfig.host.json b/packages/api/workspace-controller/tsconfig.host.json index ff1ca31080..e9d5104e98 100644 --- a/packages/api/workspace-controller/tsconfig.host.json +++ b/packages/api/workspace-controller/tsconfig.host.json @@ -21,6 +21,7 @@ { "path": "../../host/directory-picker" }, { "path": "../../storage/storage-domain" }, { "path": "../../typert/protocol" }, + { "path": "../remote-access/tsconfig.host.json" }, { "path": "../../util/deque" }, { "path": "../../util/native-command" }, { "path": "../../workspace/workspace" } diff --git a/packages/goal/goal/tests/goal.spec.ts b/packages/goal/goal/tests/goal.spec.ts index fe9c2d0030..5bcdcf1fcb 100644 --- a/packages/goal/goal/tests/goal.spec.ts +++ b/packages/goal/goal/tests/goal.spec.ts @@ -14,6 +14,7 @@ import GoalService, { } from '@deepseek-ai/dsh-goal' import type { GoalChangeMeta, GoalRef, GoalSnapshotChangeMeta } from '@deepseek-ai/dsh-goal' import { createInboxStub } from '@deepseek-ai/dsh-agent-loop-testkit' +import { remoteMethods } from '@deepseek-ai/dsh-typert-protocol' declare module '@deepseek-ai/dsh-llm' { interface MessageSourceMap { @@ -181,6 +182,16 @@ describe('GoalService creation and replay', () => { }) }) + it('declares no resource readers on goals/get, leaving its Session resource to the Agent lookup', async () => { + const ctx = new Context() + await ctx.plugin(AgentRegistry) + await ctx.plugin(SessionProjectionRegistry) + const goals = new GoalService(ctx) + const marker = remoteMethods(goals).find(method => method.method === 'get') + expect(marker).toMatchObject({ method: 'get' }) + expect(marker?.resources).toBeUndefined() + }) + it('rejects invalid direct configuration', async () => { const ctx = new Context() await ctx.plugin(AgentRegistry)