feat(api): resolve endpoint resources and filter lists by access

This commit is contained in:
imccyu committed 2026-10-13 23:01:38 +08:00
1 parent d7bd68ee7a
commit 2640d3df82
21 files changed
+527 -21

No files matched your search

@@ -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
@@ -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<WorkspaceFollowFrame, { readonly type: 'replace' }>
/**
* 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<WorkspaceFollowFrame> {
async *follow(signal: AbortSignal): AsyncIterable<WorkspaceFeedFrame> {
signal.throwIfAborted()
const follower = new WorkspaceFollower()
this.followers.add(follower)
@@ -138,7 +141,7 @@ export class WorkspaceFeed {
})
}
private publish(frame: Exclude<WorkspaceFollowFrame, { readonly type: 'baseline' }>): void {
private publish(frame: Exclude<WorkspaceFeedFrame, { readonly type: 'baseline' }>): 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<WorkspaceFollowFrame>()
private readonly frames = new Deque<WorkspaceFeedFrame>()
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<WorkspaceFollowFrame> {
async *read(signal: AbortSignal): AsyncIterable<WorkspaceFeedFrame> {
while (!this.closed && !signal.aborted) {
const frame = this.frames.popFront()
if (frame !== undefined) {
+72 -4
View File
@@ -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<Config, ResolvedConfig> = 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<WorkspaceFollowFrame> {
return this.feed.follow(signal)
async *follow(signal: AbortSignal): AsyncIterable<WorkspaceFollowFrame> {
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<WorkspaceId>()
/** 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
@@ -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[] }