feat: workspace source-of-truth hard cut

This commit is contained in:
Mohamed Boudra
2026-03-03 23:28:16 +07:00
parent f3796e2ba5
commit cb92f39e73
20 changed files with 1813 additions and 945 deletions
@@ -127,6 +127,11 @@ export type DaemonEvent =
agentId: string
payload: Extract<SessionOutboundMessage, { type: 'agent_update' }>['payload']
}
| {
type: 'workspace_update'
workspaceId: string
payload: Extract<SessionOutboundMessage, { type: 'workspace_update' }>['payload']
}
| {
type: 'agent_stream'
agentId: string
@@ -287,6 +292,19 @@ export type FetchAgentsOptions = Omit<FetchAgentsRequest, 'type' | 'requestId'>
}
export type FetchAgentsEntry = FetchAgentsPayload['entries'][number]
export type FetchAgentsPageInfo = FetchAgentsPayload['pageInfo']
type FetchWorkspacesPayload = Extract<
SessionOutboundMessage,
{ type: 'fetch_workspaces_response' }
>['payload']
type FetchWorkspacesRequest = Extract<
SessionInboundMessage,
{ type: 'fetch_workspaces_request' }
>
export type FetchWorkspacesOptions = Omit<FetchWorkspacesRequest, 'type' | 'requestId'> & {
requestId?: string
}
export type FetchWorkspacesEntry = FetchWorkspacesPayload['entries'][number]
export type FetchWorkspacesPageInfo = FetchWorkspacesPayload['pageInfo']
export type FetchAgentResult = {
agent: AgentSnapshotPayload
@@ -1098,6 +1116,33 @@ export class DaemonClient {
})
}
async fetchWorkspaces(options?: FetchWorkspacesOptions): Promise<FetchWorkspacesPayload> {
const resolvedRequestId = this.createRequestId(options?.requestId)
const message = SessionInboundMessageSchema.parse({
type: 'fetch_workspaces_request',
requestId: resolvedRequestId,
...(options?.filter ? { filter: options.filter } : {}),
...(options?.sort ? { sort: options.sort } : {}),
...(options?.page ? { page: options.page } : {}),
...(options?.subscribe ? { subscribe: options.subscribe } : {}),
})
return this.sendRequest({
requestId: resolvedRequestId,
message,
timeout: 10000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== 'fetch_workspaces_response') {
return null
}
if (msg.payload.requestId !== resolvedRequestId) {
return null
}
return msg.payload
},
})
}
async fetchAgent(agentId: string, requestId?: string): Promise<FetchAgentResult | null> {
const resolvedRequestId = this.createRequestId(requestId)
const message = SessionInboundMessageSchema.parse({
@@ -3077,6 +3122,13 @@ export class DaemonClient {
agentId: msg.payload.kind === 'upsert' ? msg.payload.agent.id : msg.payload.agentId,
payload: msg.payload,
}
case 'workspace_update':
return {
type: 'workspace_update',
workspaceId:
msg.payload.kind === 'upsert' ? msg.payload.workspace.id : msg.payload.id,
payload: msg.payload,
}
case 'agent_stream':
return {
type: 'agent_stream',
+549 -19
View File
@@ -30,6 +30,8 @@ import {
type DirectorySuggestionsRequest,
type ProjectCheckoutLitePayload,
type ProjectPlacementPayload,
type WorkspaceDescriptorPayload,
type WorkspaceStateBucket,
} from './messages.js'
import type { TerminalManager, TerminalsChangedEvent } from '../terminal/terminal-manager.js'
import type { TerminalSession } from '../terminal/terminal.js'
@@ -301,6 +303,31 @@ type FetchAgentsCursor = {
values: Record<string, string | number | null>
id: string
}
type FetchWorkspacesRequestMessage = Extract<
SessionInboundMessage,
{ type: 'fetch_workspaces_request' }
>
type FetchWorkspacesRequestFilter = NonNullable<FetchWorkspacesRequestMessage['filter']>
type FetchWorkspacesRequestSort = NonNullable<FetchWorkspacesRequestMessage['sort']>[number]
type FetchWorkspacesResponsePayload = Extract<
SessionOutboundMessage,
{ type: 'fetch_workspaces_response' }
>['payload']
type FetchWorkspacesResponseEntry = FetchWorkspacesResponsePayload['entries'][number]
type FetchWorkspacesResponsePageInfo = FetchWorkspacesResponsePayload['pageInfo']
type WorkspaceUpdatePayload = Extract<SessionOutboundMessage, { type: 'workspace_update' }>['payload']
type WorkspaceUpdatesFilter = FetchWorkspacesRequestFilter
type WorkspaceUpdatesSubscriptionState = {
subscriptionId: string
filter?: WorkspaceUpdatesFilter
isBootstrapping: boolean
pendingUpdatesByWorkspaceId: Map<string, WorkspaceUpdatePayload>
}
type FetchWorkspacesCursor = {
sort: FetchWorkspacesRequestSort[]
values: Record<string, string | number | null>
id: string
}
class SessionRequestError extends Error {
constructor(
@@ -552,6 +579,7 @@ export class Session {
private readonly providerRegistry: ReturnType<typeof buildProviderRegistry>
private unsubscribeAgentEvents: (() => void) | null = null
private agentUpdatesSubscription: AgentUpdatesSubscriptionState | null = null
private workspaceUpdatesSubscription: WorkspaceUpdatesSubscriptionState | null = null
private clientActivity: {
deviceType: 'web' | 'mobile'
focusedAgentId: string | null
@@ -1258,31 +1286,30 @@ export class Session {
private async forwardAgentUpdate(agent: ManagedAgent): Promise<void> {
try {
const subscription = this.agentUpdatesSubscription
if (!subscription) {
return
}
const payload = await this.buildAgentPayload(agent)
const project = await this.buildProjectPlacement(payload.cwd)
const matches = this.matchesAgentFilter({
agent: payload,
project,
filter: subscription.filter,
})
if (matches) {
this.bufferOrEmitAgentUpdate(subscription, {
kind: 'upsert',
if (subscription) {
const project = await this.buildProjectPlacement(payload.cwd)
const matches = this.matchesAgentFilter({
agent: payload,
project,
filter: subscription.filter,
})
return
if (matches) {
this.bufferOrEmitAgentUpdate(subscription, {
kind: 'upsert',
agent: payload,
project,
})
} else {
this.bufferOrEmitAgentUpdate(subscription, {
kind: 'remove',
agentId: payload.id,
})
}
}
this.bufferOrEmitAgentUpdate(subscription, {
kind: 'remove',
agentId: payload.id,
})
await this.emitWorkspaceUpdateForCwd(payload.cwd)
} catch (error) {
this.sessionLogger.error({ err: error }, 'Failed to emit agent update')
}
@@ -1310,6 +1337,10 @@ export class Session {
await this.handleFetchAgents(msg)
break
case 'fetch_workspaces_request':
await this.handleFetchWorkspacesRequest(msg)
break
case 'fetch_agent_request':
await this.handleFetchAgent(msg.agentId, msg.requestId)
break
@@ -1728,6 +1759,11 @@ export class Session {
private async handleDeleteAgentRequest(agentId: string, requestId: string): Promise<void> {
this.sessionLogger.info({ agentId }, `Deleting agent ${agentId} from registry`)
const knownCwd =
this.agentManager.getAgent(agentId)?.cwd ??
(await this.agentStorage.get(agentId))?.cwd ??
null
// Prevent the persistence hook from re-creating the record while we close/delete.
this.agentStorage.beginDelete(agentId)
@@ -1763,6 +1799,10 @@ export class Session {
agentId,
})
}
if (knownCwd) {
await this.emitWorkspaceUpdateForCwd(knownCwd)
}
}
private async handleArchiveAgentRequest(agentId: string, requestId: string): Promise<void> {
@@ -5203,6 +5243,434 @@ export class Session {
}
}
private readonly workspaceStatePriority: Record<WorkspaceStateBucket, number> = {
needs_input: 0,
failed: 1,
running: 2,
attention: 3,
done: 4,
}
private normalizeWorkspaceId(cwd: string): string {
const trimmed = cwd.trim()
if (!trimmed) {
return cwd
}
return resolve(trimmed)
}
private deriveWorkspaceStateBucket(agent: AgentSnapshotPayload): WorkspaceStateBucket {
const pendingPermissionCount = agent.pendingPermissions?.length ?? 0
if (pendingPermissionCount > 0 || agent.attentionReason === 'permission') {
return 'needs_input'
}
if (agent.status === 'error' || agent.attentionReason === 'error') {
return 'failed'
}
if (agent.status === 'running' || agent.status === 'initializing') {
return 'running'
}
if (agent.requiresAttention) {
return 'attention'
}
return 'done'
}
private deriveWorkspaceDirectoryName(cwd: string): string {
const normalized = cwd.replace(/\\/g, '/')
const segments = normalized.split('/').filter(Boolean)
return segments[segments.length - 1] ?? cwd
}
private deriveWorkspaceName(input: {
cwd: string
checkout: ProjectCheckoutLitePayload
}): string {
const branch = input.checkout.currentBranch?.trim() ?? null
if (branch && branch.toUpperCase() !== 'HEAD') {
return branch
}
return this.deriveWorkspaceDirectoryName(input.cwd)
}
private accumulateLatestActivityAt(
current: string | null,
agent: AgentSnapshotPayload
): string | null {
const candidateRaw = agent.lastUserMessageAt ?? agent.updatedAt
const candidateMs = Date.parse(candidateRaw)
if (Number.isNaN(candidateMs)) {
return current
}
if (!current) {
return new Date(candidateMs).toISOString()
}
const currentMs = Date.parse(current)
if (Number.isNaN(currentMs) || candidateMs > currentMs) {
return new Date(candidateMs).toISOString()
}
return current
}
private async listWorkspaceDescriptors(): Promise<WorkspaceDescriptorPayload[]> {
const agents = await this.listAgentPayloads({
labels: { ui: 'true' },
})
const descriptorsByWorkspaceId = new Map<string, WorkspaceDescriptorPayload>()
const placementByWorkspaceId = new Map<string, Promise<ProjectPlacementPayload>>()
const getPlacement = (workspaceCwd: string): Promise<ProjectPlacementPayload> => {
const key = this.normalizeWorkspaceId(workspaceCwd)
const existing = placementByWorkspaceId.get(key)
if (existing) {
return existing
}
const next = this.buildProjectPlacement(workspaceCwd)
placementByWorkspaceId.set(key, next)
return next
}
for (const agent of agents) {
if (agent.archivedAt) {
continue
}
const workspaceId = this.normalizeWorkspaceId(agent.cwd)
const placement = await getPlacement(workspaceId)
const existing = descriptorsByWorkspaceId.get(workspaceId)
if (!existing) {
const bucket = this.deriveWorkspaceStateBucket(agent)
descriptorsByWorkspaceId.set(workspaceId, {
id: workspaceId,
projectId: placement.projectKey,
name: this.deriveWorkspaceName({
cwd: workspaceId,
checkout: placement.checkout,
}),
status: bucket,
activityAt: this.accumulateLatestActivityAt(null, agent),
})
continue
}
const bucket = this.deriveWorkspaceStateBucket(agent)
if (this.workspaceStatePriority[bucket] < this.workspaceStatePriority[existing.status]) {
existing.status = bucket
}
existing.activityAt = this.accumulateLatestActivityAt(existing.activityAt, agent)
}
return Array.from(descriptorsByWorkspaceId.values())
}
private normalizeFetchWorkspacesSort(
sort: FetchWorkspacesRequestSort[] | undefined
): FetchWorkspacesRequestSort[] {
const fallback: FetchWorkspacesRequestSort[] = [{ key: 'activity_at', direction: 'desc' }]
if (!sort || sort.length === 0) {
return fallback
}
const deduped: FetchWorkspacesRequestSort[] = []
const seen = new Set<string>()
for (const entry of sort) {
if (seen.has(entry.key)) {
continue
}
seen.add(entry.key)
deduped.push(entry)
}
return deduped.length > 0 ? deduped : fallback
}
private getFetchWorkspacesSortValue(
workspace: WorkspaceDescriptorPayload,
key: FetchWorkspacesRequestSort['key']
): string | number | null {
switch (key) {
case 'status_priority':
return this.workspaceStatePriority[workspace.status]
case 'activity_at':
return workspace.activityAt ? Date.parse(workspace.activityAt) : null
case 'name':
return workspace.name.toLocaleLowerCase()
case 'project_id':
return workspace.projectId.toLocaleLowerCase()
}
}
private compareFetchWorkspacesEntries(
left: WorkspaceDescriptorPayload,
right: WorkspaceDescriptorPayload,
sort: FetchWorkspacesRequestSort[]
): number {
for (const spec of sort) {
const leftValue = this.getFetchWorkspacesSortValue(left, spec.key)
const rightValue = this.getFetchWorkspacesSortValue(right, spec.key)
const base = this.compareSortValues(leftValue, rightValue)
if (base === 0) {
continue
}
return spec.direction === 'asc' ? base : -base
}
return left.id.localeCompare(right.id)
}
private encodeFetchWorkspacesCursor(
entry: FetchWorkspacesResponseEntry,
sort: FetchWorkspacesRequestSort[]
): string {
const values: Record<string, string | number | null> = {}
for (const spec of sort) {
values[spec.key] = this.getFetchWorkspacesSortValue(entry, spec.key)
}
return Buffer.from(
JSON.stringify({
sort,
values,
id: entry.id,
}),
'utf8'
).toString('base64url')
}
private decodeFetchWorkspacesCursor(
cursor: string,
sort: FetchWorkspacesRequestSort[]
): FetchWorkspacesCursor {
let parsed: unknown
try {
parsed = JSON.parse(Buffer.from(cursor, 'base64url').toString('utf8'))
} catch {
throw new SessionRequestError('invalid_cursor', 'Invalid fetch_workspaces cursor')
}
if (!parsed || typeof parsed !== 'object') {
throw new SessionRequestError('invalid_cursor', 'Invalid fetch_workspaces cursor')
}
const payload = parsed as {
sort?: unknown
values?: unknown
id?: unknown
}
if (!Array.isArray(payload.sort) || typeof payload.id !== 'string') {
throw new SessionRequestError('invalid_cursor', 'Invalid fetch_workspaces cursor')
}
if (!payload.values || typeof payload.values !== 'object') {
throw new SessionRequestError('invalid_cursor', 'Invalid fetch_workspaces cursor')
}
const cursorSort: FetchWorkspacesRequestSort[] = []
for (const item of payload.sort) {
if (
!item ||
typeof item !== 'object' ||
typeof (item as { key?: unknown }).key !== 'string' ||
typeof (item as { direction?: unknown }).direction !== 'string'
) {
throw new SessionRequestError('invalid_cursor', 'Invalid fetch_workspaces cursor')
}
const key = (item as { key: string }).key
const direction = (item as { direction: string }).direction
if (
(key !== 'status_priority' &&
key !== 'activity_at' &&
key !== 'name' &&
key !== 'project_id') ||
(direction !== 'asc' && direction !== 'desc')
) {
throw new SessionRequestError('invalid_cursor', 'Invalid fetch_workspaces cursor')
}
cursorSort.push({ key, direction })
}
if (
cursorSort.length !== sort.length ||
cursorSort.some(
(entry, index) =>
entry.key !== sort[index]?.key || entry.direction !== sort[index]?.direction
)
) {
throw new SessionRequestError(
'invalid_cursor',
'fetch_workspaces cursor does not match current sort'
)
}
return {
sort: cursorSort,
values: payload.values as Record<string, string | number | null>,
id: payload.id,
}
}
private compareWorkspaceWithCursor(
workspace: WorkspaceDescriptorPayload,
cursor: FetchWorkspacesCursor,
sort: FetchWorkspacesRequestSort[]
): number {
for (const spec of sort) {
const leftValue = this.getFetchWorkspacesSortValue(workspace, spec.key)
const rightValue =
cursor.values[spec.key] !== undefined ? (cursor.values[spec.key] ?? null) : null
const base = this.compareSortValues(leftValue, rightValue)
if (base === 0) {
continue
}
return spec.direction === 'asc' ? base : -base
}
return workspace.id.localeCompare(cursor.id)
}
private matchesWorkspaceFilter(input: {
workspace: WorkspaceDescriptorPayload
filter: FetchWorkspacesRequestFilter | undefined
}): boolean {
const { workspace, filter } = input
if (!filter) {
return true
}
if (filter.projectId && filter.projectId.trim().length > 0) {
if (workspace.projectId !== filter.projectId.trim()) {
return false
}
}
if (filter.idPrefix && filter.idPrefix.trim().length > 0) {
if (!workspace.id.startsWith(filter.idPrefix.trim())) {
return false
}
}
if (filter.query && filter.query.trim().length > 0) {
const query = filter.query.trim().toLocaleLowerCase()
const haystacks = [workspace.name, workspace.projectId, workspace.id]
if (!haystacks.some((value) => value.toLocaleLowerCase().includes(query))) {
return false
}
}
return true
}
private async listFetchWorkspacesEntries(
request: Extract<SessionInboundMessage, { type: 'fetch_workspaces_request' }>
): Promise<{
entries: FetchWorkspacesResponseEntry[]
pageInfo: FetchWorkspacesResponsePageInfo
}> {
const filter = request.filter
const sort = this.normalizeFetchWorkspacesSort(request.sort)
let entries = await this.listWorkspaceDescriptors()
entries = entries.filter((workspace) => this.matchesWorkspaceFilter({ workspace, filter }))
entries.sort((left, right) => this.compareFetchWorkspacesEntries(left, right, sort))
const cursorToken = request.page?.cursor
if (cursorToken) {
const cursor = this.decodeFetchWorkspacesCursor(cursorToken, sort)
entries = entries.filter((workspace) => this.compareWorkspaceWithCursor(workspace, cursor, sort) > 0)
}
const limit = request.page?.limit ?? 200
const pagedEntries = entries.slice(0, limit)
const hasMore = entries.length > limit
const nextCursor =
hasMore && pagedEntries.length > 0
? this.encodeFetchWorkspacesCursor(pagedEntries[pagedEntries.length - 1], sort)
: null
return {
entries: pagedEntries,
pageInfo: {
nextCursor,
prevCursor: request.page?.cursor ?? null,
hasMore,
},
}
}
private bufferOrEmitWorkspaceUpdate(
subscription: WorkspaceUpdatesSubscriptionState,
payload: WorkspaceUpdatePayload
): void {
if (subscription.isBootstrapping) {
const workspaceId = payload.kind === 'upsert' ? payload.workspace.id : payload.id
subscription.pendingUpdatesByWorkspaceId.set(workspaceId, payload)
return
}
this.emit({
type: 'workspace_update',
payload,
})
}
private flushBootstrappedWorkspaceUpdates(options?: {
snapshotLatestActivityByWorkspaceId?: Map<string, number>
}): void {
const subscription = this.workspaceUpdatesSubscription
if (!subscription || !subscription.isBootstrapping) {
return
}
subscription.isBootstrapping = false
const pending = Array.from(subscription.pendingUpdatesByWorkspaceId.values())
subscription.pendingUpdatesByWorkspaceId.clear()
for (const payload of pending) {
if (payload.kind === 'upsert') {
const snapshotLatestActivity = options?.snapshotLatestActivityByWorkspaceId?.get(
payload.workspace.id
)
if (typeof snapshotLatestActivity === 'number') {
const updateLatestActivity = payload.workspace.activityAt
? Date.parse(payload.workspace.activityAt)
: Number.NEGATIVE_INFINITY
if (!Number.isNaN(updateLatestActivity) && updateLatestActivity <= snapshotLatestActivity) {
continue
}
}
}
this.emit({
type: 'workspace_update',
payload,
})
}
}
private async emitWorkspaceUpdateForCwd(cwd: string): Promise<void> {
const subscription = this.workspaceUpdatesSubscription
if (!subscription) {
return
}
const workspaceId = this.normalizeWorkspaceId(cwd)
const all = await this.listWorkspaceDescriptors()
const workspace = all.find((entry) => entry.id === workspaceId)
if (!workspace) {
this.bufferOrEmitWorkspaceUpdate(subscription, {
kind: 'remove',
id: workspaceId,
})
return
}
if (!this.matchesWorkspaceFilter({ workspace, filter: subscription.filter })) {
this.bufferOrEmitWorkspaceUpdate(subscription, {
kind: 'remove',
id: workspaceId,
})
return
}
this.bufferOrEmitWorkspaceUpdate(subscription, {
kind: 'upsert',
workspace,
})
}
private async handleFetchAgents(
request: Extract<SessionInboundMessage, { type: 'fetch_agents_request' }>
): Promise<void> {
@@ -5263,6 +5731,68 @@ export class Session {
}
}
private async handleFetchWorkspacesRequest(
request: Extract<SessionInboundMessage, { type: 'fetch_workspaces_request' }>
): Promise<void> {
const requestedSubscriptionId = request.subscribe?.subscriptionId?.trim()
const subscriptionId = request.subscribe
? requestedSubscriptionId && requestedSubscriptionId.length > 0
? requestedSubscriptionId
: uuidv4()
: null
try {
if (subscriptionId) {
this.workspaceUpdatesSubscription = {
subscriptionId,
filter: request.filter,
isBootstrapping: true,
pendingUpdatesByWorkspaceId: new Map(),
}
}
const payload = await this.listFetchWorkspacesEntries(request)
const snapshotLatestActivityByWorkspaceId = new Map<string, number>()
for (const entry of payload.entries) {
const parsedLatestActivity = entry.activityAt
? Date.parse(entry.activityAt)
: Number.NEGATIVE_INFINITY
if (!Number.isNaN(parsedLatestActivity)) {
snapshotLatestActivityByWorkspaceId.set(entry.id, parsedLatestActivity)
}
}
this.emit({
type: 'fetch_workspaces_response',
payload: {
requestId: request.requestId,
...(subscriptionId ? { subscriptionId } : {}),
...payload,
},
})
if (subscriptionId && this.workspaceUpdatesSubscription?.subscriptionId === subscriptionId) {
this.flushBootstrappedWorkspaceUpdates({ snapshotLatestActivityByWorkspaceId })
}
} catch (error) {
if (subscriptionId && this.workspaceUpdatesSubscription?.subscriptionId === subscriptionId) {
this.workspaceUpdatesSubscription = null
}
const code = error instanceof SessionRequestError ? error.code : 'fetch_workspaces_failed'
const message = error instanceof Error ? error.message : 'Failed to fetch workspaces'
this.sessionLogger.error({ err: error }, 'Failed to handle fetch_workspaces_request')
this.emit({
type: 'rpc_error',
payload: {
requestId: request.requestId,
requestType: request.type,
error: message,
code,
},
})
}
}
private async handleFetchAgent(agentIdOrIdentifier: string, requestId: string): Promise<void> {
const resolved = await this.resolveAgentIdentifier(agentIdOrIdentifier)
if (!resolved.ok) {
@@ -0,0 +1,267 @@
import { describe, expect, test, vi } from 'vitest'
import { Session } from './session.js'
import type { AgentSnapshotPayload } from '../shared/messages.js'
function makeAgent(input: {
id: string
cwd: string
status: AgentSnapshotPayload['status']
updatedAt: string
pendingPermissions?: number
requiresAttention?: boolean
attentionReason?: AgentSnapshotPayload['attentionReason']
}): AgentSnapshotPayload {
const pendingPermissionCount = input.pendingPermissions ?? 0
return {
id: input.id,
provider: 'codex',
cwd: input.cwd,
model: null,
thinkingOptionId: null,
effectiveThinkingOptionId: null,
createdAt: input.updatedAt,
updatedAt: input.updatedAt,
lastUserMessageAt: null,
status: input.status,
capabilities: {
supportsStreaming: true,
supportsSessionPersistence: true,
supportsDynamicModes: true,
supportsMcpServers: true,
supportsReasoningStream: true,
supportsToolInvocations: true,
},
currentModeId: null,
availableModes: [],
pendingPermissions: Array.from({ length: pendingPermissionCount }, (_, index) => ({
id: `perm-${input.id}-${index}`,
provider: 'codex',
name: 'tool',
kind: 'tool',
})),
persistence: null,
runtimeInfo: {
provider: 'codex',
sessionId: null,
},
title: null,
labels: { ui: 'true' },
requiresAttention: input.requiresAttention ?? false,
attentionReason: input.attentionReason ?? null,
attentionTimestamp: null,
archivedAt: null,
}
}
function createSessionForWorkspaceTests(): Session {
const logger = {
child: () => logger,
trace: vi.fn(),
debug: vi.fn(),
info: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
}
return new Session({
clientId: 'test-client',
onMessage: vi.fn(),
logger: logger as any,
downloadTokenStore: {} as any,
pushTokenStore: {} as any,
paseoHome: '/tmp/paseo-test',
agentManager: {
subscribe: () => () => {},
listAgents: () => [],
getAgent: () => null,
} as any,
agentStorage: {
list: async () => [],
get: async () => null,
} as any,
createAgentMcpTransport: async () => {
throw new Error('not used')
},
stt: null,
tts: null,
terminalManager: null,
})
}
describe('workspace aggregation', () => {
test('non-git workspace uses deterministic directory name and no unknown branch fallback', async () => {
const session = createSessionForWorkspaceTests() as any
session.listAgentPayloads = async () => [
makeAgent({
id: 'a1',
cwd: '/tmp/non-git',
status: 'idle',
updatedAt: '2026-03-01T12:00:00.000Z',
}),
]
session.buildProjectPlacement = async (cwd: string) => ({
projectKey: cwd,
projectName: 'non-git',
checkout: {
cwd,
isGit: false,
currentBranch: null,
remoteUrl: null,
isPaseoOwnedWorktree: false,
mainRepoRoot: null,
},
})
const result = await session.listFetchWorkspacesEntries({
type: 'fetch_workspaces_request',
requestId: 'req-1',
})
expect(result.entries).toHaveLength(1)
expect(result.entries[0]?.name).toBe('non-git')
expect(result.entries[0]?.name).not.toBe('Unknown branch')
})
test('git branch workspace uses branch as canonical name', async () => {
const session = createSessionForWorkspaceTests() as any
session.listAgentPayloads = async () => [
makeAgent({
id: 'a1',
cwd: '/tmp/repo-branch',
status: 'running',
updatedAt: '2026-03-01T12:00:00.000Z',
}),
]
session.buildProjectPlacement = async (cwd: string) => ({
projectKey: cwd,
projectName: 'repo-branch',
checkout: {
cwd,
isGit: true,
currentBranch: 'feature/name-from-server',
remoteUrl: 'https://github.com/acme/repo-branch.git',
isPaseoOwnedWorktree: false,
mainRepoRoot: null,
},
})
const result = await session.listFetchWorkspacesEntries({
type: 'fetch_workspaces_request',
requestId: 'req-branch',
})
expect(result.entries).toHaveLength(1)
expect(result.entries[0]?.name).toBe('feature/name-from-server')
})
test('branch/detached policies and dominant status bucket are deterministic', async () => {
const session = createSessionForWorkspaceTests() as any
session.listAgentPayloads = async () => [
makeAgent({
id: 'a1',
cwd: '/tmp/repo',
status: 'running',
updatedAt: '2026-03-01T12:00:00.000Z',
}),
makeAgent({
id: 'a2',
cwd: '/tmp/repo',
status: 'error',
updatedAt: '2026-03-01T12:01:00.000Z',
}),
makeAgent({
id: 'a3',
cwd: '/tmp/repo',
status: 'idle',
updatedAt: '2026-03-01T12:02:00.000Z',
pendingPermissions: 1,
}),
]
session.buildProjectPlacement = async (cwd: string) => ({
projectKey: cwd,
projectName: 'repo',
checkout: {
cwd,
isGit: true,
currentBranch: 'HEAD',
remoteUrl: 'https://github.com/acme/repo.git',
isPaseoOwnedWorktree: false,
mainRepoRoot: null,
},
})
const result = await session.listFetchWorkspacesEntries({
type: 'fetch_workspaces_request',
requestId: 'req-2',
})
expect(result.entries).toHaveLength(1)
expect(result.entries[0]?.name).toBe('repo')
expect(result.entries[0]?.status).toBe('needs_input')
})
test('workspace update stream emits upsert and remove on lifecycle changes', async () => {
const emitted: Array<{ type: string; payload: unknown }> = []
const logger = {
child: () => logger,
trace: vi.fn(),
debug: vi.fn(),
info: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
}
const session = new Session({
clientId: 'test-client',
onMessage: (message) => emitted.push(message as any),
logger: logger as any,
downloadTokenStore: {} as any,
pushTokenStore: {} as any,
paseoHome: '/tmp/paseo-test',
agentManager: {
subscribe: () => () => {},
listAgents: () => [],
getAgent: () => null,
} as any,
agentStorage: {
list: async () => [],
get: async () => null,
} as any,
createAgentMcpTransport: async () => {
throw new Error('not used')
},
stt: null,
tts: null,
terminalManager: null,
}) as any
session.workspaceUpdatesSubscription = {
subscriptionId: 'sub-1',
filter: undefined,
isBootstrapping: false,
pendingUpdatesByWorkspaceId: new Map(),
}
session.listWorkspaceDescriptors = async () => [
{
id: '/tmp/repo',
projectId: '/tmp/repo',
name: 'repo',
status: 'running',
activityAt: '2026-03-01T12:00:00.000Z',
},
]
await session.emitWorkspaceUpdateForCwd('/tmp/repo')
session.listWorkspaceDescriptors = async () => []
await session.emitWorkspaceUpdateForCwd('/tmp/repo')
const workspaceUpdates = emitted.filter((message) => message.type === 'workspace_update')
expect(workspaceUpdates).toHaveLength(2)
expect((workspaceUpdates[0] as any).payload.kind).toBe('upsert')
expect((workspaceUpdates[1] as any).payload).toEqual({
kind: 'remove',
id: '/tmp/repo',
})
})
})
+82
View File
@@ -512,6 +512,45 @@ export const FetchAgentsRequestMessageSchema = z.object({
.optional(),
})
const WorkspaceStateBucketSchema = z.enum([
'needs_input',
'failed',
'running',
'attention',
'done',
])
export const FetchWorkspacesRequestMessageSchema = z.object({
type: z.literal('fetch_workspaces_request'),
requestId: z.string(),
filter: z
.object({
query: z.string().optional(),
projectId: z.string().optional(),
idPrefix: z.string().optional(),
})
.optional(),
sort: z
.array(
z.object({
key: z.enum(['status_priority', 'activity_at', 'name', 'project_id']),
direction: z.enum(['asc', 'desc']),
})
)
.optional(),
page: z
.object({
limit: z.number().int().positive().max(200),
cursor: z.string().min(1).optional(),
})
.optional(),
subscribe: z
.object({
subscriptionId: z.string().optional(),
})
.optional(),
})
export const FetchAgentRequestMessageSchema = z.object({
type: z.literal('fetch_agent_request'),
requestId: z.string(),
@@ -1068,6 +1107,7 @@ export const SessionInboundMessageSchema = z.discriminatedUnion('type', [
AbortRequestMessageSchema,
AudioPlayedMessageSchema,
FetchAgentsRequestMessageSchema,
FetchWorkspacesRequestMessageSchema,
FetchAgentRequestMessageSchema,
DeleteAgentRequestMessageSchema,
ArchiveAgentRequestMessageSchema,
@@ -1426,6 +1466,14 @@ export const ProjectPlacementPayloadSchema = z.object({
checkout: ProjectCheckoutLitePayloadSchema,
})
export const WorkspaceDescriptorPayloadSchema = z.object({
id: z.string(),
projectId: z.string(),
name: z.string(),
status: WorkspaceStateBucketSchema,
activityAt: z.string().nullable(),
})
export const AgentUpdateMessageSchema = z.object({
type: z.literal('agent_update'),
payload: z.discriminatedUnion('kind', [
@@ -1488,6 +1536,34 @@ export const FetchAgentsResponseMessageSchema = z.object({
}),
})
export const FetchWorkspacesResponseMessageSchema = z.object({
type: z.literal('fetch_workspaces_response'),
payload: z.object({
requestId: z.string(),
subscriptionId: z.string().nullable().optional(),
entries: z.array(WorkspaceDescriptorPayloadSchema),
pageInfo: z.object({
nextCursor: z.string().nullable(),
prevCursor: z.string().nullable(),
hasMore: z.boolean(),
}),
}),
})
export const WorkspaceUpdateMessageSchema = z.object({
type: z.literal('workspace_update'),
payload: z.discriminatedUnion('kind', [
z.object({
kind: z.literal('upsert'),
workspace: WorkspaceDescriptorPayloadSchema,
}),
z.object({
kind: z.literal('remove'),
id: z.string(),
}),
]),
})
export const FetchAgentResponseMessageSchema = z.object({
type: z.literal('fetch_agent_response'),
payload: z.object({
@@ -2051,9 +2127,11 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion('type', [
RpcErrorMessageSchema,
ArtifactMessageSchema,
AgentUpdateMessageSchema,
WorkspaceUpdateMessageSchema,
AgentStreamMessageSchema,
AgentStatusMessageSchema,
FetchAgentsResponseMessageSchema,
FetchWorkspacesResponseMessageSchema,
FetchAgentResponseMessageSchema,
FetchAgentTimelineResponseMessageSchema,
SendAgentMessageResponseMessageSchema,
@@ -2119,7 +2197,10 @@ export type AgentStreamMessage = z.infer<typeof AgentStreamMessageSchema>
export type AgentStatusMessage = z.infer<typeof AgentStatusMessageSchema>
export type ProjectCheckoutLitePayload = z.infer<typeof ProjectCheckoutLitePayloadSchema>
export type ProjectPlacementPayload = z.infer<typeof ProjectPlacementPayloadSchema>
export type WorkspaceStateBucket = z.infer<typeof WorkspaceStateBucketSchema>
export type WorkspaceDescriptorPayload = z.infer<typeof WorkspaceDescriptorPayloadSchema>
export type FetchAgentsResponseMessage = z.infer<typeof FetchAgentsResponseMessageSchema>
export type FetchWorkspacesResponseMessage = z.infer<typeof FetchWorkspacesResponseMessageSchema>
export type FetchAgentResponseMessage = z.infer<typeof FetchAgentResponseMessageSchema>
export type FetchAgentTimelineResponseMessage = z.infer<
typeof FetchAgentTimelineResponseMessageSchema
@@ -2144,6 +2225,7 @@ export type ActivityLogPayload = z.infer<typeof ActivityLogPayloadSchema>
// Type exports for inbound message types
export type VoiceAudioChunkMessage = z.infer<typeof VoiceAudioChunkMessageSchema>
export type FetchAgentsRequestMessage = z.infer<typeof FetchAgentsRequestMessageSchema>
export type FetchWorkspacesRequestMessage = z.infer<typeof FetchWorkspacesRequestMessageSchema>
export type FetchAgentRequestMessage = z.infer<typeof FetchAgentRequestMessageSchema>
export type SendAgentMessageRequest = z.infer<typeof SendAgentMessageRequestSchema>
export type WaitForFinishRequest = z.infer<typeof WaitForFinishRequestSchema>
@@ -0,0 +1,42 @@
import { describe, expect, test } from 'vitest'
import {
SessionInboundMessageSchema,
SessionOutboundMessageSchema,
} from './messages.js'
describe('workspace message schemas', () => {
test('parses fetch_workspaces_request', () => {
const parsed = SessionInboundMessageSchema.parse({
type: 'fetch_workspaces_request',
requestId: 'req-1',
filter: {
query: 'repo',
projectId: 'remote:github.com/acme/repo',
idPrefix: '/Users/me',
},
sort: [{ key: 'activity_at', direction: 'desc' }],
page: { limit: 50 },
subscribe: {},
})
expect(parsed.type).toBe('fetch_workspaces_request')
})
test('rejects invalid workspace update payload', () => {
const result = SessionOutboundMessageSchema.safeParse({
type: 'workspace_update',
payload: {
kind: 'upsert',
workspace: {
id: '/repo',
projectId: '/repo',
name: '',
status: 'not-a-bucket',
activityAt: null,
},
},
})
expect(result.success).toBe(false)
})
})