diff --git a/cli/README.md b/cli/README.md index 7866fcad..af9e1ad5 100644 --- a/cli/README.md +++ b/cli/README.md @@ -32,7 +32,7 @@ Run Claude Code, Codex, Cursor Agent, DeepSeek Harness, Grok Build, or OpenCode Supports `hapi cursor resume `, `hapi cursor --continue`, `--mode plan|ask`, `--yolo`, `--model`. Local and remote modes supported; remote uses `agent -p` with stream-json. - `hapi dsh` - Start or resume a DeepSeek Harness session through `dsh web`. - Start DSH first with `dsh web --port 3080`. HAPI uses `http://127.0.0.1:3080` by default; set `HAPI_DSH_URL` to override it. + Start DSH first with `dsh web --port 3080`. HAPI uses `http://127.0.0.1:3080` by default; set `HAPI_DSH_URL` to override it. DSH 0.2.x launch tokens can be supplied with `HAPI_DSH_TOKEN` or `HAPI_DSH_TOKEN_FILE` (the bundled launch wrapper writes this file). - `hapi grok` - Start Grok Build mode. See `src/grok/runGrok.ts`. - `hapi opencode` - Start OpenCode mode via ACP. See `src/opencode/runOpencode.ts`. Note: OpenCode supports local and remote modes; local mode streams via OpenCode plugins. diff --git a/cli/src/agent/agentAvailability.ts b/cli/src/agent/agentAvailability.ts index a163e291..291c535e 100644 --- a/cli/src/agent/agentAvailability.ts +++ b/cli/src/agent/agentAvailability.ts @@ -108,7 +108,21 @@ async function getDshWebAvailability( ): Promise { let client: DshWebClient try { - client = new DshWebClient(env.HAPI_DSH_URL, fetchImpl) + // Do not let an availability probe with an injected environment pick + // up the current process user's token file. The normal process.env + // path keeps DSH's default token discovery; isolated probes retain + // only the token/file explicitly present in their environment. + const explicitToken = env.HAPI_DSH_TOKEN + const explicitTokenFile = env.HAPI_DSH_TOKEN_FILE + const ambientEnvironment = env === process.env + client = new DshWebClient( + env.HAPI_DSH_URL, + fetchImpl, + explicitToken !== undefined || explicitTokenFile !== undefined + ? explicitToken + : ambientEnvironment ? undefined : '', + ambientEnvironment ? undefined : explicitTokenFile + ) } catch { return { agent: 'dsh', available: false, reason: 'invalid_configuration' } } diff --git a/cli/src/claude/utils/startHappyServer.ts b/cli/src/claude/utils/startHappyServer.ts index 80d7089f..70c60039 100644 --- a/cli/src/claude/utils/startHappyServer.ts +++ b/cli/src/claude/utils/startHappyServer.ts @@ -36,6 +36,7 @@ import { PermissionModeSchema } from '@hapi/protocol/schemas' import { PingPeerError, formatInspectPeerReport, formatPeerSessionsList, inspectPeer, listPeerSessions, peerListFetchLimit, pingPeer } from "@/modules/pingPeer/pingPeer"; import { TeamClientError, + createTeamTask, formatTeamMessages, formatTeamStatus, formatTeamTasks, @@ -640,7 +641,7 @@ function createHapiMcpServer( }); mcp.registerTool('team_send', { - description: 'Agent Team: send a message to teammates. to="all" (default) writes to the shared log only; to="" or to="lead" wakes that member; to="human" notifies the human out-of-band (no inbox item); use kind="decision" when you need the human to decide something. Use inReplyTo= when answering a peer message so the hub can stop runaway back-and-forth. Do NOT use this for routine replies to the human.', + description: 'Agent Team: send a message to teammates. to="all" (default) writes to the shared log only; to="lead" wakes the Lead; non-Lead members can only direct-message the Lead (peer-to-peer DMs are blocked); to="human" notifies the human out-of-band (no inbox item); use kind="decision" when you need the human to decide something. Use inReplyTo= when answering a peer message so the hub can stop runaway back-and-forth. Do NOT use this for routine replies to the human.', title: 'Send Team Message', inputSchema: z.object({ text: z.string().min(1).describe('Message text'), @@ -715,21 +716,44 @@ function createHapiMcpServer( }); mcp.registerTool('team_task', { - description: 'Agent Team: list team tasks or update one. Moving a task to doing/done requires its dependencies to be done; marking a task done requires a deliverable (evidence: branch/commit/files/test result).', + description: 'Agent Team: list team tasks, create one for an existing member (Lead only), or update one. Moving a task to doing/done requires its dependencies to be done; marking a task done requires a deliverable (evidence: branch/commit/files/test result). A member may hold only one doing task at a time.', title: 'Team Tasks', inputSchema: z.object({ - action: z.enum(['list', 'update']).describe('list = show team tasks; update = change one task'), + action: z.enum(['list', 'create', 'update']).describe('list = show team tasks; create = add a task for an existing member (Lead only); update = change one task'), taskId: z.string().min(1).optional().describe('Required for action=update'), - status: z.enum(['todo', 'doing', 'done', 'blocked']).optional().describe('New status'), + title: z.string().min(1).max(200).optional().describe('Required for action=create'), + assigneeSessionId: z.string().min(1).nullable().optional().describe('Assignee session id (create)'), + status: z.enum(['todo', 'doing', 'done', 'blocked']).optional().describe('New status (update)'), deliverable: z.string().min(1).max(2000).optional().describe('Evidence for done: branch/commit/files/test result'), dependsOn: z.array(z.string().min(1)).max(20).optional().describe('Task ids that must be done first'), + requirementId: z.string().min(1).optional().describe('Requirement the new task belongs to (create)'), }), - }, async (args: { action: 'list' | 'update'; taskId?: string; status?: 'todo' | 'doing' | 'done' | 'blocked'; deliverable?: string; dependsOn?: string[] }) => { + }, async (args: { action: 'list' | 'create' | 'update'; taskId?: string; title?: string; assigneeSessionId?: string | null; status?: 'todo' | 'doing' | 'done' | 'blocked'; deliverable?: string; dependsOn?: string[]; requirementId?: string }) => { try { const status = await requireTeam(); if (args.action === 'list') { return { content: [{ type: 'text' as const, text: formatTeamTasks(status.tasks, client.sessionId) }], isError: false }; } + if (args.action === 'create') { + if (!args.title) { + return { content: [{ type: 'text' as const, text: 'title is required for action=create' }], isError: true }; + } + const task = await createTeamTask({ + sessionId: client.sessionId, + teamId: status.team.id, + title: args.title, + assigneeSessionId: args.assigneeSessionId ?? null, + dependsOn: args.dependsOn, + requirementId: args.requirementId + }); + return { + content: [{ + type: 'text' as const, + text: `已创建任务「${task.title}」 id=${task.id}${task.assigneeSessionId ? '，已指派并通知成员' : ''}` + }], + isError: false + }; + } if (!args.taskId) { return { content: [{ type: 'text' as const, text: 'taskId is required for action=update' }], isError: true }; } diff --git a/cli/src/codex/happyMcpStdioBridge.ts b/cli/src/codex/happyMcpStdioBridge.ts index 175fccb2..9425b4fb 100644 --- a/cli/src/codex/happyMcpStdioBridge.ts +++ b/cli/src/codex/happyMcpStdioBridge.ts @@ -393,7 +393,7 @@ export async function runHappyMcpStdioBridge(argv: string[]): Promise { { name: 'team_send', title: 'Send Team Message', - description: 'Agent Team: send a message to teammates. to="all" (default) writes to the shared log only; to="" or to="lead" wakes that member; to="human" notifies the human out-of-band (no inbox item); use kind="decision" when you need the human to decide something. Use inReplyTo= when answering a peer message.', + description: 'Agent Team: send a message to teammates. to="all" (default) writes to the shared log only; to="lead" wakes the Lead; non-Lead members can only direct-message the Lead (peer-to-peer DMs are blocked); to="human" notifies the human out-of-band (no inbox item); use kind="decision" when you need the human to decide something. Use inReplyTo= when answering a peer message.', inputSchema: z.object({ text: z.string().min(1).describe('Message text'), to: z.string().min(1).optional().describe('"all" (default), "lead", "human", or a member session id/prefix'), @@ -422,13 +422,16 @@ export async function runHappyMcpStdioBridge(argv: string[]): Promise { { name: 'team_task', title: 'Team Tasks', - description: 'Agent Team: list team tasks or update one. doing/done requires dependencies to be done; done requires a deliverable (evidence).', + description: 'Agent Team: list team tasks, create one for an existing member (Lead only), or update one. doing/done requires dependencies to be done; done requires a deliverable (evidence). A member may hold only one doing task at a time.', inputSchema: z.object({ - action: z.enum(['list', 'update']), + action: z.enum(['list', 'create', 'update']), taskId: z.string().min(1).optional().describe('Required for action=update'), + title: z.string().min(1).max(200).optional().describe('Required for action=create'), + assigneeSessionId: z.string().min(1).nullable().optional().describe('Assignee session id (create)'), status: z.enum(['todo', 'doing', 'done', 'blocked']).optional(), deliverable: z.string().min(1).max(2000).optional().describe('Evidence for done: branch/commit/files/test result'), dependsOn: z.array(z.string().min(1)).max(20).optional(), + requirementId: z.string().min(1).optional().describe('Requirement the new task belongs to (create)'), }), }, { diff --git a/cli/src/dsh/dshWebClient.test.ts b/cli/src/dsh/dshWebClient.test.ts index d1e74601..4e6f6511 100644 --- a/cli/src/dsh/dshWebClient.test.ts +++ b/cli/src/dsh/dshWebClient.test.ts @@ -40,7 +40,7 @@ describe('DshWebClient', () => { }] }] } - })) + }), '') await expect(client.getModels('session-1')).resolves.toEqual({ current: { provider: 'deepseek-official', model: 'deepseek-v4-pro', reasoningEffort: 'max' }, @@ -60,7 +60,7 @@ describe('DshWebClient', () => { it('keeps DSH RPC errors structured', async () => { const client = new DshWebClient('http://127.0.0.1:3080', rpcFetch({ 'host.describe': new Error('not ready') - })) + }), '') const error = await client.describe().catch((value) => value) expect(error).toBeInstanceOf(DshWebRpcError) expect(error).toMatchObject({ method: 'host.describe', code: 'failed' }) @@ -82,7 +82,7 @@ describe('DshWebClient', () => { rpcId: request.rpcId, result: { ok: true, value } }), { status: 200, headers: { 'content-type': 'application/json' } }) - }) as typeof fetch) + }) as typeof fetch, '') await expect(client.executeCommand('session-1', '/compact')).resolves.toEqual({ commandId: 'cmd-1', @@ -95,7 +95,7 @@ describe('DshWebClient', () => { }) it('returns undefined for a DSH line that is not a native command', async () => { - const client = new DshWebClient('http://127.0.0.1:3080', rpcFetch({ 'commands/execute': undefined })) + const client = new DshWebClient('http://127.0.0.1:3080', rpcFetch({ 'commands/execute': undefined }), '') await expect(client.executeCommand('session-1', '/unknown')).resolves.toBeUndefined() }) @@ -105,7 +105,7 @@ describe('DshWebClient', () => { commandId: 'cmd-2', result: { kind: 'error', text: 'Compaction is unavailable because the agent is busy.' } } - })) + }), '') await expect(client.executeCommand('session-1', '/compact')).resolves.toEqual({ commandId: 'cmd-2', @@ -145,7 +145,7 @@ describe('DshWebClient', () => { }] }] } - })) + }), '') await expect(listDshModels(client)).resolves.toEqual({ success: true, @@ -163,4 +163,41 @@ describe('DshWebClient', () => { }] }) }) + + it('authenticates a 0.2.x host and sends generated Remote args', async () => { + const requests: Array<{ url: string; body?: Record; headers?: Headers }> = [] + const client = new DshWebClient('http://127.0.0.1:3080', (async (input, init) => { + const url = String(input) + requests.push({ + url, + body: init?.body ? JSON.parse(String(init.body)) as Record : undefined, + headers: new Headers(init?.headers) + }) + if (url.endsWith('/connect?token=test-token&interactions=answer')) { + return new Response(null, { + status: 302, + headers: { 'set-cookie': 'dsh-auth-test=abc; Path=/; HttpOnly' } + }) + } + return new Response(JSON.stringify({ + type: 'server-response', + rpcId: requests.at(-1)?.body?.rpcId, + result: { + ok: true, + value: { + items: [{ sessionId: 's1', updatedAt: 1, running: false, blank: false }] + } + } + }), { status: 200 }) + }) as typeof fetch, 'test-token') + + await expect(client.listSessions()).resolves.toEqual([{ + sessionId: 's1', updatedAt: 1, running: false, blank: false + }]) + expect(requests[1].body).toMatchObject({ + method: 'session/list', + payload: { args: { _request: {} } } + }) + expect(requests[1].headers?.get('cookie')).toBe('dsh-auth-test=abc') + }) }) diff --git a/cli/src/dsh/dshWebClient.ts b/cli/src/dsh/dshWebClient.ts index ecb173d0..2524624a 100644 --- a/cli/src/dsh/dshWebClient.ts +++ b/cli/src/dsh/dshWebClient.ts @@ -1,4 +1,7 @@ import { randomUUID } from 'node:crypto' +import { readFileSync } from 'node:fs' +import { homedir } from 'node:os' +import { join } from 'node:path' type JsonRecord = Record @@ -26,6 +29,7 @@ export type DshSessionSummary = { origin?: 'subagent' cwd?: string agentPreset?: string + agentAvailable?: boolean projections?: { asOfSeq: number values: Record @@ -87,13 +91,69 @@ export function resolveDshWebUrl(value = process.env.HAPI_DSH_URL): string { if (url.protocol !== 'http:' && url.protocol !== 'https:') { throw new Error('HAPI_DSH_URL must use http:// or https://') } - if (url.username || url.password || url.search || url.hash) { - throw new Error('HAPI_DSH_URL must not contain credentials, query parameters, or a fragment') + if (url.username || url.password || url.hash) { + throw new Error('HAPI_DSH_URL must not contain credentials or a fragment') } + if ([...url.searchParams.keys()].some((key) => key !== 'token')) { + throw new Error('HAPI_DSH_URL only accepts the DSH launch token query parameter') + } + url.search = '' url.pathname = '/' return url.origin } +/** + * DSH 0.2.x prints a short-lived launch token in its web URL. Keep the URL + * parser strict, but accept the printed URL at the client boundary and strip + * its token before constructing the base URL. + */ +function resolveConfiguredUrl(value: string | undefined): { url: string; token?: string } { + const candidate = value?.trim() || 'http://127.0.0.1:3080' + const parsed = new URL(candidate) + const token = parsed.searchParams.get('token') || undefined + if (token) parsed.search = '' + return { url: resolveDshWebUrl(parsed.toString()), ...(token ? { token } : {}) } +} + +function tokenFromFile(path: string): string | undefined { + try { + const text = readFileSync(path, 'utf8').trim() + if (!text) return undefined + try { + const url = new URL(text) + return url.searchParams.get('token') || undefined + } catch { + return text.split(/\s+/u)[0] + } + } catch { + return undefined + } +} + +function configuredToken(explicit?: string, explicitFile?: string): string | undefined { + // An explicit empty token disables file discovery. This keeps callers + // that inject a fetch implementation (tests and availability probes) + // isolated from a developer's real ~/.dsh token. + if (explicit !== undefined) return explicit.trim() || undefined + if (explicitFile !== undefined) { + const token = tokenFromFile(explicitFile.trim()) + return token + } + if (process.env.HAPI_DSH_TOKEN?.trim()) return process.env.HAPI_DSH_TOKEN.trim() + const configuredPath = process.env.HAPI_DSH_TOKEN_FILE?.trim() + const paths = configuredPath + ? [configuredPath] + : [ + join(homedir(), '.dsh', 'hapi-token'), + join(homedir(), '.local', 'share', 'deepseek-harness', 'state', 'hapi-token') + ] + for (const path of paths) { + const token = tokenFromFile(path) + if (token) return token + } + return undefined +} + function isRecord(value: unknown): value is JsonRecord { return typeof value === 'object' && value !== null && !Array.isArray(value) } @@ -149,18 +209,28 @@ function parseSessionSummary(value: unknown): DshSessionSummary | null { ...(value.origin === 'subagent' ? { origin: 'subagent' as const } : {}), ...(asString(value.cwd) ? { cwd: value.cwd as string } : {}), ...(asString(value.agentPreset) ? { agentPreset: value.agentPreset as string } : {}), + ...(typeof value.agentAvailable === 'boolean' ? { agentAvailable: value.agentAvailable } : {}), ...(projections ? { projections } : {}) } } export class DshWebClient { readonly baseUrl: string + private readonly launchToken?: string + private authCookie?: string + private remoteEventCookie?: string + private authMode: 'unknown' | 'legacy' | 'remote' = 'unknown' + private remoteClientId?: string constructor( - baseUrl = resolveDshWebUrl(), - private readonly fetchImpl: typeof fetch = fetch + baseUrl = process.env.HAPI_DSH_URL ?? 'http://127.0.0.1:3080', + private readonly fetchImpl: typeof fetch = fetch, + token?: string, + tokenFile?: string ) { - this.baseUrl = resolveDshWebUrl(baseUrl) + const configured = resolveConfiguredUrl(baseUrl) + this.baseUrl = configured.url + this.launchToken = configured.token ?? configuredToken(token, tokenFile) } async describe(signal?: AbortSignal): Promise<{ @@ -170,6 +240,18 @@ export class DshWebClient { model: string attachedSessions: number }> { + await this.ensureAuth(signal) + if (this.authMode === 'remote') { + const sessions = await this.listSessions(signal) + const cwd = sessions.find((entry) => entry.cwd)?.cwd ?? process.cwd() + return { + version: '0.2.1', + cwd, + provider: 'unknown', + model: 'unknown', + attachedSessions: sessions.filter((entry) => entry.agentAvailable ?? entry.running).length + } + } const { value } = await this.call('host.describe', {}, signal) if (!isRecord(value)) throw new Error('DeepSeek Harness host.describe returned an invalid value') const version = asString(value.version) @@ -206,12 +288,32 @@ export class DshWebClient { ...(options.beforeSeq !== undefined ? { beforeSeq: options.beforeSeq } : {}), ...(options.maxMessages !== undefined ? { maxMessages: options.maxMessages } : {}) } - const { value } = await this.call('session.history', payload, options.signal, 120_000) - if (!isRecord(value) || !Array.isArray(value.events) || typeof value.hasMore !== 'boolean') { - throw new Error('DeepSeek Harness session.history returned an invalid value') + const remote = await this.ensureAuth(options.signal) + const throughSeq = remote + ? await this.readSessionCursor(options.sessionId, options.signal) + : undefined + const { value } = remote + ? await this.call('session.page', { + request: { + address: { kind: 'session', sessionId: options.sessionId }, + // A page is inclusive. The follow opening snapshot gives + // us the valid current tail cursor. + throughSeq, + ...(options.beforeSeq !== undefined ? { beforeSeq: options.beforeSeq } : {}), + ...(options.maxMessages !== undefined ? { maxMessages: options.maxMessages } : {}) + } + }, options.signal, 120_000) + : await this.call('session.history', payload, options.signal, 120_000) + const rows = isRecord(value) && Array.isArray(value.events) + ? value.events + : isRecord(value) && Array.isArray(value.records) + ? value.records + : null + if (!isRecord(value) || !rows || typeof value.hasMore !== 'boolean') { + throw new Error(`DeepSeek Harness ${remote ? 'session.page' : 'session.history'} returned an invalid value`) } const entries: DshHistoryEntry[] = [] - for (const raw of value.events) { + for (const raw of rows) { if (!isRecord(raw)) throw new Error('DeepSeek Harness session.history returned an invalid entry') const event = parseSessionEvent(raw.event) if (!event) throw new Error('DeepSeek Harness session.history returned an invalid event') @@ -242,6 +344,7 @@ export class DshWebClient { signal?: AbortSignal }): Promise<{ rpcId: string; command?: { text?: string } }> { const { rpcId, value } = await this.call('session.prompt', { + requestId: options.rpcId ?? randomUUID(), sessionId: options.sessionId, mode: options.mode ?? 'queue', content: [{ type: 'text', text: options.text }], @@ -314,12 +417,35 @@ export class DshWebClient { current: DshModelSelection models: DshModelSummary[] }> { - const { value } = await this.call('session.models', { sessionId }, signal) - if (!isRecord(value) || !isRecord(value.current) || !Array.isArray(value.groups)) { + const remote = await this.ensureAuth(signal) + const { value } = remote + ? await this.call('session.modelCatalog', {}, signal) + : await this.call('session.models', { sessionId }, signal) + if (!isRecord(value) || !Array.isArray(value.groups)) { throw new Error('DeepSeek Harness session.models returned an invalid value') } - const provider = asString(value.current.provider) - const model = asString(value.current.model) + let currentValue = remote + ? (isRecord(value.default) ? value.default : null) + : (isRecord(value.current) ? value.current : null) + if (remote) { + const projections = await this.call('session/projections', { + request: { sessionId } + }, signal) + const projectionValue = isRecord(projections.value) && isRecord(projections.value.values) + ? projections.value.values + : null + const selection = projectionValue && isRecord(projectionValue.modelSelection) + ? (isRecord(projectionValue.modelSelection.next) + ? projectionValue.modelSelection.next + : isRecord(projectionValue.modelSelection.lastUsed) + ? projectionValue.modelSelection.lastUsed + : null) + : null + if (selection) currentValue = selection + } + if (!currentValue) throw new Error('DeepSeek Harness session.models is missing the current model') + const provider = asString(currentValue.provider) + const model = asString(currentValue.model) if (!provider || !model) { throw new Error('DeepSeek Harness session.models is missing the current model') } @@ -361,8 +487,8 @@ export class DshWebClient { current: { provider, model, - ...(asString(value.current.reasoningEffort) - ? { reasoningEffort: value.current.reasoningEffort as string } + ...(asString(currentValue.reasoningEffort) + ? { reasoningEffort: currentValue.reasoningEffort as string } : {}) }, models @@ -418,6 +544,52 @@ export class DshWebClient { } private async sendResponse(rpcId: string, result: unknown, signal?: AbortSignal): Promise { + if (await this.ensureAuth(signal)) { + const unwrapped = isRecord(result) && result.ok === true && Object.hasOwn(result, 'value') + ? result.value + : result + const outcome = isRecord(result) && result.ok === false + ? { + kind: 'rejected', + error: { + name: 'AbortError', + message: isRecord(result.error) && asString(result.error.message) + ? result.error.message + : 'Cancelled by HAPI client' + } + } + : isRecord(unwrapped) && typeof unwrapped.outcome === 'string' + ? { kind: 'result', value: unwrapped.outcome } + : isRecord(unwrapped) && unwrapped.answer !== undefined + ? { kind: 'result', value: unwrapped.answer } + : { kind: 'result', value: unwrapped } + const response = await this.fetchImpl(new URL('/api/$events/result', this.baseUrl), { + method: 'POST', + headers: { + ...(this.remoteEventCookie ? { cookie: this.remoteEventCookie } : this.headers()), + 'content-type': 'application/json' + }, + body: JSON.stringify({ + type: 'client-request', + rpcId: randomUUID(), + method: '$events/result', + payload: { + args: { + clientId: this.remoteClientId, + eventId: rpcId, + outcome + } + } + }), + signal + }) + if (!response.ok) throw new Error(`DeepSeek Harness response transport failed: HTTP ${response.status}`) + const body: unknown = await response.json() + if (!isRecord(body) || !isRecord(body.result) || body.result.ok !== true) { + throw new Error('DeepSeek Harness response was rejected') + } + return + } const response = await this.fetchImpl(new URL('/api/respond', this.baseUrl), { method: 'POST', headers: { 'content-type': 'application/json' }, @@ -441,6 +613,9 @@ export class DshWebClient { onFrame: (request: DshServerRequest) => void onError?: (error: Error) => void }): Promise { + if (this.authMode === 'remote' || this.launchToken) { + return this.subscribeRemoteEvents(options) + } const url = new URL('/api/events.mux', this.baseUrl) url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:' return new Promise((resolve, reject) => { @@ -487,6 +662,303 @@ export class DshWebClient { }) } + /** Open one latest DSH session.follow stream and project its entries into + * the legacy event callback shape used by HAPI's runner. */ + subscribeSession(options: { + sessionId: string + signal: AbortSignal + onEvent: (event: DshSessionEvent, view?: unknown) => void + onSnapshot?: (records: DshHistoryEntry[], cursor?: number) => void + onError?: (error: Error) => void + }): Promise { + // DSH binds one API cookie to one mux peer. Use a dedicated client for + // the long-lived follow stream so this instance can keep using its + // cookie for unary prompt/model/cancel calls without starving the + // stream of live event frames. + const streamClient = new DshWebClient(this.baseUrl, this.fetchImpl, this.launchToken) + return streamClient.ensureAuth(options.signal).then((remote) => { + if (!remote) return + return streamClient.openSessionStream(options) + }) + } + + private openSessionStream(options: { + sessionId: string + signal: AbortSignal + onEvent: (event: DshSessionEvent, view?: unknown) => void + onSnapshot?: (records: DshHistoryEntry[], cursor?: number) => void + onError?: (error: Error) => void + }): Promise { + return this.subscribeRemoteStream({ + endpoint: 'session/follow', + payload: { + args: { + request: { + address: { kind: 'session', sessionId: options.sessionId }, + assistantStream: true + } + } + }, + signal: options.signal, + onValue: (value) => { + if (!isRecord(value)) return + if (value.type === 'snapshot' && Array.isArray(value.records)) { + const records = value.records.flatMap((record): DshHistoryEntry[] => { + if (!isRecord(record)) return [] + const event = parseSessionEvent(record.event) + return event + ? [{ event, ...(record.view !== undefined ? { view: record.view } : {}) }] + : [] + }) + options.onSnapshot?.( + records, + asNumber(value.cursor) ?? undefined + ) + for (const record of records) options.onEvent(record.event, record.view) + } else if (value.type === 'entry' || value.type === 'event') { + this.emitFollowRecord(value, options.onEvent) + } + }, + onError: options.onError + }) + } + + private emitFollowRecord(value: unknown, onEvent: (event: DshSessionEvent, view?: unknown) => void): void { + if (!isRecord(value)) return + const event = parseSessionEvent(value.event) + if (event) onEvent(event, value.view) + } + + private subscribeRemoteEvents(options: { + signal: AbortSignal + onFrame: (request: DshServerRequest) => void + onError?: (error: Error) => void + }): Promise { + const rememberCookie = this.ensureAuth(options.signal).then((remote) => { + if (remote) this.remoteEventCookie = this.authCookie + return remote + }) + return rememberCookie.then((remote) => remote ? this.subscribeRemoteStream({ + endpoint: '$events', + payload: { args: {} }, + signal: options.signal, + onValue: (value) => { + if (!isRecord(value)) return + if (value.type === 'ready' && asString(value.clientId)) { + this.remoteClientId = value.clientId as string + return + } + if (value.type === 'waterfall' && asString(value.eventId) && asString(value.event) + && asString(value.agentId) && isRecord(value.request)) { + const event = value.event as string + const request = value.request + if (event === 'approval/request' || event === 'approval/requested') { + options.onFrame({ + type: 'server-request', + rpcId: value.eventId as string, + method: event, + payload: { + type: 'approval/requested', + sessionId: value.agentId as string, + approvalId: value.eventId, + ...(asString(request.toolName) ? { toolName: request.toolName } : {}), + ...(asString(request.callId) ? { callId: request.callId } : {}), + ...(asString(request.reason) ? { reason: request.reason } : {}) + } + }) + } else if (event === 'user-questions/request' || event === 'question/requested') { + options.onFrame({ + type: 'server-request', + rpcId: value.eventId as string, + method: event, + payload: { + type: 'question/requested', + sessionId: value.agentId as string, + questions: Array.isArray(request.questions) ? request.questions : [] + } + }) + } + return + } + // A few 0.2.x builds expose a broadcasted session/event on the + // Remote Event stream. Accept both [session,event] and a + // single object form without making it a dependency. + if (value.type === 'emit' && value.event === 'session/event' && Array.isArray(value.args)) { + const candidate = value.args.at(-1) + const event = parseSessionEvent(candidate) + const sessionId = value.args.length > 1 && asString(value.args[0]) ? value.args[0] : undefined + if (event) { + options.onFrame({ + type: 'server-request', + rpcId: randomUUID(), + method: 'session/event', + payload: { type: 'session/event', ...(sessionId ? { sessionId } : {}), event } + }) + } + } + }, + onError: options.onError + }) : Promise.resolve()) + } + + private async readSessionCursor(sessionId: string, signal?: AbortSignal): Promise { + if (this.remoteEventCookie && this.authCookie === this.remoteEventCookie) { + this.authMode = 'unknown' + this.authCookie = undefined + await this.ensureAuth(signal) + } + const controller = new AbortController() + const abort = () => controller.abort(signal?.reason) + signal?.addEventListener('abort', abort, { once: true }) + let cursor = -1 + try { + await this.subscribeRemoteStream({ + endpoint: 'session/follow', + payload: { + args: { + request: { + address: { kind: 'session', sessionId } + } + } + }, + signal: controller.signal, + waitForFirst: true, + onValue: (value) => { + if (isRecord(value) && value.type === 'snapshot' && asNumber(value.cursor) !== null) { + cursor = value.cursor as number + } + } + }) + } finally { + signal?.removeEventListener('abort', abort) + controller.abort() + // DSH binds an API cookie to the lifetime of its mux peer. Closing + // this short-lived cursor probe retires that peer, so exchange the + // launch token once more before the following unary page request. + if (!signal?.aborted) { + this.authMode = 'unknown' + this.authCookie = undefined + await this.ensureAuth(signal) + } + } + return cursor + } + + private subscribeRemoteStream(options: { + endpoint: string + payload: unknown + signal: AbortSignal + onValue: (value: unknown) => void + onError?: (error: Error) => void + waitForFirst?: boolean + }): Promise { + return this.ensureAuth(options.signal).then(() => new Promise((resolve, reject) => { + const url = new URL('/api/remote.mux', this.baseUrl) + url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:' + const WebSocketCtor = WebSocket as unknown as new ( + url: string, + options?: { headers?: Record } + ) => WebSocket + const socket = new WebSocketCtor(url.toString(), this.authCookie ? { headers: { cookie: this.authCookie } } : undefined) + const streamId = randomUUID() + let opened = false + let firstItem = false + let ended = false + const abort = () => { + if (socket.readyState === WebSocket.OPEN) { + socket.send(JSON.stringify({ type: 'cancel', streamId })) + } + socket.close() + } + options.signal.addEventListener('abort', abort, { once: true }) + socket.onopen = () => { + opened = true + socket.send(JSON.stringify({ + type: 'open', + streamId, + endpoint: options.endpoint, + payload: options.payload + })) + if (!options.waitForFirst) resolve() + } + socket.onmessage = (message) => { + try { + const parsed: unknown = JSON.parse(String(message.data)) + if (!isRecord(parsed) || parsed.streamId !== streamId) return + if (parsed.type === 'item') { + options.onValue(parsed.value) + if (options.waitForFirst && !firstItem) { + firstItem = true + resolve() + } + } else if (parsed.type === 'error') { + const errorValue = isRecord(parsed.error) ? parsed.error : {} + throw new Error(asString(errorValue.message) ?? 'DeepSeek Harness remote stream failed') + } else if (parsed.type === 'end') { + ended = true + socket.close() + } + } catch (error) { + const failure = error instanceof Error ? error : new Error(String(error)) + if (!opened) reject(failure) + else options.onError?.(failure) + } + } + socket.onerror = () => { + const error = new Error('DeepSeek Harness remote stream failed') + if (!opened) reject(error) + else options.onError?.(error) + } + socket.onclose = () => { + options.signal.removeEventListener('abort', abort) + if (!options.signal.aborted && !ended) { + const error = new Error('DeepSeek Harness remote stream disconnected') + if (!opened) reject(error) + else options.onError?.(error) + } + } + })) + } + + private headers(extra?: Record): Record { + return { + ...(this.authCookie ? { cookie: this.authCookie } : {}), + ...(extra ?? {}) + } + } + + private async ensureAuth(signal?: AbortSignal): Promise { + if (this.authMode === 'remote') return true + if (this.authMode === 'legacy') return false + if (!this.launchToken) { + this.authMode = 'legacy' + return false + } + const url = new URL('/connect', this.baseUrl) + url.searchParams.set('token', this.launchToken) + url.searchParams.set('interactions', 'answer') + let response: Response + try { + response = await this.fetchImpl(url, { + method: 'GET', + redirect: 'manual', + signal + }) + } catch (error) { + throw new Error(`Unable to authenticate with DeepSeek Harness at ${this.baseUrl}: ${error instanceof Error ? error.message : String(error)}`) + } + if (response.status < 300 || response.status >= 400) { + throw new Error(`DeepSeek Harness authentication failed: HTTP ${response.status}`) + } + const setCookie = typeof response.headers.getSetCookie === 'function' + ? response.headers.getSetCookie()[0] + : response.headers.get('set-cookie') + if (!setCookie) throw new Error('DeepSeek Harness authentication did not return an API cookie') + this.authCookie = setCookie.split(';', 1)[0] + this.authMode = 'remote' + return true + } + private async call( method: string, payload: unknown, @@ -499,10 +971,41 @@ export class DshWebClient { const combined = signal ? AbortSignal.any([timeout, signal]) : timeout let response: Response try { - response = await this.fetchImpl(new URL(`/api/${method}`, this.baseUrl), { + const remote = await this.ensureAuth(combined) + // The generated Remote gateway reserves one `args` object. Keep + // the old wire shape when no launch token is configured so HAPI + // remains usable with a 0.1.x host during a rolling upgrade. + const args = remote + ? method === 'session.list' && isRecord(payload) && Object.keys(payload).length === 0 + ? { _request: {} } + : ['session.create', 'session.prompt', 'session.cancel', 'session.selectModel'].includes(method) + ? isRecord(payload) && Object.keys(payload).length === 1 && isRecord(payload.request) + ? payload + : { request: payload } + : method === 'commands/execute' && isRecord(payload) && isRecord(payload.args) + ? { + ...(isRecord(payload.args) && asString(payload.args.agentId) + ? { agentId: payload.args.agentId } + : {}), + ...(isRecord(payload.args) && typeof payload.args.line === 'string' + ? { line: payload.args.line } + : {}), + submittedAttachments: isRecord(payload.args) && Array.isArray(payload.args.images) + ? payload.args.images + : [] + } + : payload + : payload + const wireMethod = remote ? method.replace('.', '/') : method + response = await this.fetchImpl(new URL(`/api/${wireMethod}`, this.baseUrl), { method: 'POST', - headers: { 'content-type': 'application/json' }, - body: JSON.stringify({ type: 'client-request', rpcId, method, payload }), + headers: this.headers({ 'content-type': 'application/json' }), + body: JSON.stringify({ + type: 'client-request', + rpcId, + method: wireMethod, + payload: remote ? { args } : payload + }), signal: combined }) } catch (error) { diff --git a/cli/src/dsh/runDsh.ts b/cli/src/dsh/runDsh.ts index 933964c2..31cf4508 100644 --- a/cli/src/dsh/runDsh.ts +++ b/cli/src/dsh/runDsh.ts @@ -265,6 +265,10 @@ export async function runDsh(opts: { await client.describe() const muxAbort = new AbortController() + // The latest DSH protocol exposes session state through a dedicated + // `session.follow` stream. Keep it separate from the interaction event + // stream so a reconnect or an approval request cannot stall event replay. + const sessionAbort = new AbortController() const loopAbort = new AbortController() const queue = new MessageQueue2((mode) => hashObject(mode)) const pendingTurns = new Map() @@ -297,6 +301,7 @@ export async function runDsh(opts: { let nativeTurnStateObserved = false let keepAliveInterval: ReturnType | null = null let latestEventSeq = sessionInfo.metadata?.dshHistoryLastEventSeq ?? -1 + let staleHistoryCursor = false const syncKeepAlive = () => { session.keepAlive(thinking, 'remote', { @@ -371,6 +376,7 @@ export async function runDsh(opts: { queue.close() loopAbort.abort() muxAbort.abort() + sessionAbort.abort() failPendingTurns(new Error('DeepSeek Harness session stopped')) finishPendingApprovals('Session stopped') finishPendingQuestions('Session stopped') @@ -511,6 +517,12 @@ export async function runDsh(opts: { if (frame.type !== 'session/event' || !frame.event) return const event = frame.event + // `session.follow` starts with a snapshot. A resumed HAPI session has + // already imported events through dshHistoryLastEventSeq; applying + // those records again would duplicate transcript messages. Sequence + // numbers are monotonic per native session, so this also makes event + // stream reconnects idempotent. + if (event.seq <= latestEventSeq) return if (event.type === 'tool/call' && isRecord(event.data) && event.data.name === 'ask_user_question' && typeof event.data.callId === 'string') { @@ -597,6 +609,22 @@ export async function runDsh(opts: { nativeSummary = (await client.listSessions()).find((entry) => entry.sessionId === nativeSessionId) if (!nativeSummary) throw new Error(`DeepSeek Harness session not found: ${nativeSessionId}`) } + // DSH 0.2.x can rewrite a session history during compaction. The + // persisted cursor then may be from the pre-compaction sequence space + // (larger than the native session's current cursor), which would make + // the event filter below discard every new event forever. Keep the + // cursor inside the current native sequence space; subsequent events + // advance it normally and the next turn-end persists the repaired + // value. + const nativeCursor = nativeSummary?.projections?.asOfSeq + if (nativeCursor !== undefined && latestEventSeq > nativeCursor) { + logger.debug('[dsh] Resetting stale history cursor after native compaction', { + persisted: latestEventSeq, + native: nativeCursor + }) + latestEventSeq = nativeCursor + staleHistoryCursor = true + } if (!nativeTurnStateObserved) thinking = nativeSummary?.running === true const projectedPermission = summaryPermissionPreset(nativeSummary?.projections?.values.permissions) @@ -607,6 +635,50 @@ export async function runDsh(opts: { session.updateMetadata((metadata) => ({ ...metadata, dshSessionId: nativeSessionId! })) + const sessionStreamError = (error: Error) => { + if (stopped || sessionAbort.signal.aborted) return + session.sendSessionEvent({ type: 'error', message: error.message }) + failPendingTurns(error) + loopAbort.abort(error) + } + await client.subscribeSession({ + sessionId: nativeSessionId, + signal: sessionAbort.signal, + onSnapshot: (records) => { + if (!staleHistoryCursor || records.length === 0) return + // After compaction, DSH's sequence space is unrelated to the + // persisted HAPI cursor. The follow snapshot includes a bounded + // tail, so recover only the latest turn's assistant output; + // user prompts and older compacted turns are already in HAPI. + let lastTurnStart = -1 + for (let index = records.length - 1; index >= 0; index -= 1) { + if (records[index]!.event.type === 'turn/start') { + lastTurnStart = index + break + } + } + const firstAssistant = lastTurnStart >= 0 + ? records.find((record, index) => index > lastTurnStart && record.event.type === 'assistant/message') + : undefined + latestEventSeq = (firstAssistant?.event.seq ?? records[lastTurnStart >= 0 ? lastTurnStart : 0]!.event.seq) - 1 + staleHistoryCursor = false + }, + onEvent: (event, view) => { + handleMuxFrame({ + type: 'server-request', + rpcId: randomUUID(), + method: 'session/event', + payload: { + type: 'session/event', + sessionId: nativeSessionId!, + event, + ...(view !== undefined ? { view } : {}) + } + }) + }, + onError: sessionStreamError + }) + const catalog = await client.getModels(nativeSessionId) currentSelection = catalog.current currentModel = currentSelection.model @@ -915,10 +987,12 @@ export async function runDsh(opts: { pendingTurns.delete(rpcId) ownedRpcIds.delete(rpcId) } - for (let index = batch.items.length - 1; index >= 0; index -= 1) { - const item = batch.items[index]! - if (batch.isolate) queue.unshiftIsolated(item.message, batch.mode, item.localId) - else queue.unshift(item.message, batch.mode, item.localId) + if (!stopped) { + for (let index = batch.items.length - 1; index >= 0; index -= 1) { + const item = batch.items[index]! + if (batch.isolate) queue.unshiftIsolated(item.message, batch.mode, item.localId) + else queue.unshift(item.message, batch.mode, item.localId) + } } throw error } diff --git a/cli/src/modules/team/teamClient.ts b/cli/src/modules/team/teamClient.ts index ee26a9ee..92a06141 100644 --- a/cli/src/modules/team/teamClient.ts +++ b/cli/src/modules/team/teamClient.ts @@ -311,6 +311,32 @@ export async function updateTeamTask( return response.task ?? null } +/** Create a task for an existing member. The hub allows this only for the Lead. */ +export async function createTeamTask( + options: TeamClientOptions & { + teamId: string + title: string + assigneeSessionId?: string | null + dependsOn?: string[] + requirementId?: string + } +): Promise { + const context = await openSession(options) + const response = await request<{ task: TeamTaskView }>( + context, + 'post', + `/api/teams/${encodeURIComponent(options.teamId)}/tasks`, + { + fromSessionId: options.sessionId, + title: options.title, + ...(options.assigneeSessionId !== undefined ? { assigneeSessionId: options.assigneeSessionId } : {}), + ...(options.dependsOn && options.dependsOn.length > 0 ? { dependsOn: options.dependsOn } : {}), + ...(options.requirementId ? { requirementId: options.requirementId } : {}) + } + ) + return response.task +} + // -------------------------------------------------------------- formatting export function formatTeamStatus(status: TeamStatusView): string { diff --git a/cli/src/modules/team/teamPrompt.ts b/cli/src/modules/team/teamPrompt.ts index 90a56e7b..5d4d5ca5 100644 --- a/cli/src/modules/team/teamPrompt.ts +++ b/cli/src/modules/team/teamPrompt.ts @@ -24,7 +24,7 @@ export function getTeamPromptBlock(toolPrefix: string, env: NodeJS.ProcessEnv = return trimIdent(` You are a member of HAPI agent team "${name}" (your role: ${role}). - Start by calling ${toolPrefix}team_status to pick up your assignment, and ${toolPrefix}team_read to pull team messages (broadcasts are not pushed to you). - - Report progress, completion, and blockers with ${toolPrefix}team_send (kind=status or task-update). Pass taskId so the update is filed under the right requirement. Batch updates; do not chat back and forth. + - Report progress, completion, and blockers with ${toolPrefix}team_send (kind=status or task-update). Always address the Lead with to="lead" — direct messages to other members are blocked. Pass taskId so the update is filed under the right requirement. Batch updates; do not chat back and forth. - Track your work with ${toolPrefix}team_task (list/update): set a task to doing when you start, blocked when stuck, and done when finished. A done task requires a deliverable (evidence: branch/commit/files/test result), and doing/done requires its dependencies to be done first. - Replies to a message the human just sent you are synced into the team group chat automatically. If your turn was triggered by anything else (a teammate message, a task, a timer) and it ends with something the human must see or decide, you MUST send it explicitly with ${toolPrefix}team_send: use kind="decision" (with to="human") only when the human has to decide something, and a normal broadcast (to="all") for progress or conclusions they merely need to see. Reserve to="human" for decisions - it interrupts them out-of-band. - Spawned members inherit your tool/model/thinking level/permission by default; only pass overrides to ${toolPrefix}spawn_peer when the human explicitly asks for a different setup. diff --git a/docs/guide/agents.md b/docs/guide/agents.md index 52d02c37..6fb596a1 100644 --- a/docs/guide/agents.md +++ b/docs/guide/agents.md @@ -263,7 +263,7 @@ reconciles usage from that ledger without duplicating it. - **Claude Code** (`hapi` / `hapi claude`) — the default and recommended flavor; local sessions wrap the native TUI, remote sessions drive the Claude Agent SDK. [Claude Code docs](https://docs.anthropic.com/en/docs/claude-code) - **Codex** (`hapi codex`) — OpenAI's Codex CLI; remote sessions talk to `codex app-server` over JSON-RPC, with a dedicated `plan` collaboration mode. [openai/codex](https://github.com/openai/codex) -- **DeepSeek Harness** (`hapi dsh`) — connects to the DSH Web runtime for native session creation, resume, model and reasoning selection, permission approvals, queued-message steering, and history import. Start `dsh web --port 3080` first; set `HAPI_DSH_URL` when it is not listening on the default `http://127.0.0.1:3080`. DSH sessions are remote-control only in HAPI because the Web runtime owns the native client. +- **DeepSeek Harness** (`hapi dsh`) — connects to the DSH Web runtime for native session creation, resume, model and reasoning selection, permission approvals, queued-message steering, and history import. Start `dsh web --port 3080` first; set `HAPI_DSH_URL` when it is not listening on the default `http://127.0.0.1:3080`. DSH 0.2.x launch tokens are read from `HAPI_DSH_TOKEN` or `HAPI_DSH_TOKEN_FILE`. DSH sessions are remote-control only in HAPI because the Web runtime owns the native client. - **GitHub Copilot** (`hapi copilot`) — Copilot CLI over ACP (`copilot --acp --stdio`). [GitHub Copilot](https://github.com/features/copilot) - **Kimi** (`hapi kimi`) — Moonshot AI's Kimi CLI over ACP (`kimi acp`). [MoonshotAI/kimi-cli](https://github.com/MoonshotAI/kimi-cli) - **OpenCode** (`hapi opencode`) — the open-source OpenCode agent over ACP (`opencode acp`). [opencode.ai](https://opencode.ai) diff --git a/hub/src/sync/usageService.ts b/hub/src/sync/usageService.ts index 81be1b5d..9ed74d08 100644 --- a/hub/src/sync/usageService.ts +++ b/hub/src/sync/usageService.ts @@ -146,7 +146,8 @@ export function getUsageSummary( store: Store, namespace: string, range: string | undefined, - timeZone: string = 'UTC' + timeZone: string = 'UTC', + options: { includeBySession?: boolean } = {} ): UsageSummaryResponse { const sessions = store.sessions.getSessionsByNamespace(namespace) // New messages are indexed on insertion. Backfill history from older hubs @@ -154,7 +155,7 @@ export function getUsageSummary( store.usage.collectSessions(sessions) const now = Date.now() - const days = range === '30d' ? 30 : range === 'all' ? null : 7 + const days = range === '30d' ? 30 : range === 'all' ? null : range === '1d' ? 1 : 7 const from = days === null ? null : now - days * 24 * 60 * 60 * 1000 // Read the durable namespace ledger rather than only live session ids; // usage rows intentionally survive session deletion. @@ -205,6 +206,7 @@ export function getUsageSummary( const daily = new Map() const byAgent = new Map() const byModel = new Map() + const bySession = new Map() const sessionsWithUsage = new Set() const cumulativePrevious = new Map() const cumulativeFingerprints = new Set() @@ -293,7 +295,11 @@ export function getUsageSummary( const modelTotals = byModel.get(modelKey) ?? emptyTotals() addTotals(modelTotals, inputTokens, outputTokens, cacheReadTokens, cacheCreationTokens) byModel.set(modelKey, modelTotals) - sessionsWithUsage.add(canonicalSession(event.sessionId)) + const sessionKey = canonicalSession(event.sessionId) + const sessionTotals = bySession.get(sessionKey) ?? emptyTotals() + addTotals(sessionTotals, inputTokens, outputTokens, cacheReadTokens, cacheCreationTokens) + bySession.set(sessionKey, sessionTotals) + sessionsWithUsage.add(sessionKey) } // Reconciliation rows are absolute per-day totals, so range filtering @@ -315,7 +321,11 @@ export function getUsageSummary( const modelTotals = byModel.get(row.model) ?? emptyTotals() addTotals(modelTotals, row.inputTokens, row.outputTokens, row.cacheReadTokens, row.cacheCreationTokens, row.requests) byModel.set(row.model, modelTotals) - sessionsWithUsage.add(canonicalSession(row.sessionId)) + const sessionKey = canonicalSession(row.sessionId) + const sessionTotals = bySession.get(sessionKey) ?? emptyTotals() + addTotals(sessionTotals, row.inputTokens, row.outputTokens, row.cacheReadTokens, row.cacheCreationTokens, row.requests) + bySession.set(sessionKey, sessionTotals) + sessionsWithUsage.add(sessionKey) } const sortBuckets = (values: Map): UsageSummaryBucket[] => Array.from(values.entries()) @@ -330,6 +340,7 @@ export function getUsageSummary( .sort((a, b) => a.key.localeCompare(b.key)), byAgent: sortBuckets(byAgent), byModel: sortBuckets(byModel), + ...(options.includeBySession ? { bySession: sortBuckets(bySession) } : {}), updatedAt: now } } diff --git a/hub/src/teams/teamService.test.ts b/hub/src/teams/teamService.test.ts index 83a34b6a..37ff14fc 100644 --- a/hub/src/teams/teamService.test.ts +++ b/hub/src/teams/teamService.test.ts @@ -190,6 +190,33 @@ describe('TeamService messaging', () => { } }) + it('blocks peer-to-peer direct messages while keeping lead reports and broadcasts open', async () => { + const { service, store, team, delivered } = setup() + store.addMember(team.id, 'sess-reviewer', 'reviewer') + try { + await expect( + service.sendMessage('sess-builder', 'alpha', { text: 'psst', to: 'sess-reviewer' }) + ).rejects.toThrow('Only the Lead') + + // Prefix targets that resolve to a peer are blocked too. + await expect( + service.sendMessage('sess-builder', 'alpha', { text: 'psst', to: 'sess-rev' }) + ).rejects.toThrow('Only the Lead') + expect(delivered).toHaveLength(0) + + // Lead -> member stays open. + await service.sendMessage('sess-lead', 'alpha', { text: 'review T1', to: 'sess-reviewer' }) + expect(delivered.at(-1)?.sessionId).toBe('sess-reviewer') + + // Member -> Lead and broadcast stay open. + await service.sendMessage('sess-builder', 'alpha', { text: 'progress', to: 'lead' }) + expect(delivered.at(-1)?.sessionId).toBe('sess-lead') + await service.sendMessage('sess-builder', 'alpha', { text: 'note to all' }) + } finally { + service.close() + } + }) + it('lets the human post into the channel and pushes directed messages', async () => { const { service, delivered } = setup() try { @@ -538,6 +565,84 @@ describe('TeamService web task management', () => { } }) + it('lets the Lead create and assign tasks for existing members', async () => { + const { service, team, delivered } = setup() + try { + const task = await service.createTaskForHuman('alpha', team.id, { + title: '补测试', + assigneeSessionId: 'sess-builder' + }, 'sess-lead') + expect(task.assigneeSessionId).toBe('sess-builder') + expect(delivered.at(-1)?.sessionId).toBe('sess-builder') + expect(delivered.at(-1)?.text).toContain('来自 lead') + + const messages = service.listMessagesForHuman('alpha', team.id) + expect(messages.some((message) => message.kind === 'task-assign' && message.fromKind === 'session')).toBe(true) + + await expect( + service.createTaskForHuman('alpha', team.id, { title: '越权' }, 'sess-builder') + ).rejects.toThrow('Only the Lead') + } finally { + service.close() + } + }) + + it('allows only one doing task per member', async () => { + const { service, team } = setup() + try { + const first = await service.createTaskForHuman('alpha', team.id, { + title: 'A', + assigneeSessionId: 'sess-builder' + }) + const second = await service.createTaskForHuman('alpha', team.id, { + title: 'B', + assigneeSessionId: 'sess-builder' + }) + await service.updateTask('sess-builder', 'alpha', first.id, { status: 'doing' }) + await expect( + service.updateTask('sess-builder', 'alpha', second.id, { status: 'doing' }) + ).rejects.toThrow('已有进行中的任务') + + await service.updateTask('sess-builder', 'alpha', first.id, { + status: 'done', + deliverable: 'A 已完成' + }) + const progressed = await service.updateTask('sess-builder', 'alpha', second.id, { status: 'doing' }) + expect(progressed.status).toBe('doing') + } finally { + service.close() + } + }) + + it('only reassigns todo tasks and briefs the new member', async () => { + const { service, store, team, delivered } = setup() + store.addMember(team.id, 'sess-reviewer', 'reviewer') + try { + const todo = await service.createTaskForHuman('alpha', team.id, { + title: '补文档', + assigneeSessionId: 'sess-builder' + }) + delivered.length = 0 + const reassigned = await service.updateTask(null, 'alpha', todo.id, { + assigneeSessionId: 'sess-reviewer' + }) + expect(reassigned.assigneeSessionId).toBe('sess-reviewer') + expect(delivered.at(-1)?.sessionId).toBe('sess-reviewer') + expect(delivered.at(-1)?.text).toContain('已指派给你') + + const doing = await service.createTaskForHuman('alpha', team.id, { + title: '在办', + assigneeSessionId: 'sess-builder' + }) + await service.updateTask(null, 'alpha', doing.id, { status: 'doing' }) + await expect( + service.updateTask(null, 'alpha', doing.id, { assigneeSessionId: 'sess-reviewer' }) + ).rejects.toThrow('Only todo tasks can be reassigned') + } finally { + service.close() + } + }) + it('updates tasks from the web without a session caller', async () => { const { service, team } = setup() try { diff --git a/hub/src/teams/teamService.ts b/hub/src/teams/teamService.ts index 4ecd5139..af73ff46 100644 --- a/hub/src/teams/teamService.ts +++ b/hub/src/teams/teamService.ts @@ -1,5 +1,6 @@ import { randomBytes } from 'node:crypto' import type { AgentFlavor, PermissionMode, SyncEvent } from '@hapi/protocol/types' +import type { TeamOverviewMemberStatus, TeamOverviewTeam } from '@hapi/protocol/apiTypes' import { TeamStore, @@ -182,6 +183,57 @@ export class TeamService { } } + /** + * Cross-team dashboard read model. Live member status uses the same + * derivation as the team page; token usage is attached by the HTTP layer. + */ + getOverview(namespace: string): TeamOverviewTeam[] { + return this.listTeams(namespace).map((team) => { + const detail = this.getTeamDetail(team.id, namespace) + const tasks = detail?.tasks ?? [] + const members = detail?.members ?? [] + const requirements = detail?.requirements ?? [] + const doingByAssignee = new Map() + const counts: Record = { todo: 0, doing: 0, blocked: 0, done: 0 } + for (const task of tasks) { + counts[task.status] += 1 + if (task.status === 'doing' && task.assigneeSessionId && !doingByAssignee.has(task.assigneeSessionId)) { + doingByAssignee.set(task.assigneeSessionId, task) + } + } + const lastBySession = this.store.lastMessageAtBySession(team.id) + const pendingDecisions = this.store.listMessages(team.id, { limit: 2000 }) + .filter(isPendingHumanDecision).length + + return { + id: team.id, + name: team.name, + status: team.status, + leadSessionId: team.leadSessionId, + members: members.map((member) => { + const currentTask = doingByAssignee.get(member.sessionId) + return { + sessionId: member.sessionId, + role: member.role, + status: member.status as TeamOverviewMemberStatus, + currentTask: currentTask ? { id: currentTask.id, title: currentTask.title } : null, + lastReportAt: lastBySession[member.sessionId] ?? null + } + }), + tasks: { total: tasks.length, ...counts }, + requirements: { + total: requirements.length, + open: requirements.filter((requirement) => requirement.status !== 'done').length, + missingConclusion: requirements.filter((requirement) => + requirement.status === 'done' && !(requirement.conclusion ?? '').trim() + ).length + }, + pendingDecisions, + lastActivityAt: this.store.lastMessageAt(team.id) + } + }) + } + createTeam(namespace: string, input: CreateTeamServiceInput): TeamRecord { const storeInput: CreateTeamInput = { namespace, @@ -262,6 +314,7 @@ export class TeamService { if (target.toSessionId === sessionId) { throw new TeamServiceError('invalid', 'Cannot send a team message to yourself') } + this.assertMemberTargetAllowed(team, sessionId, target) return await this.appendAndRoute(team, target, { fromKind: 'session', fromSessionId: sessionId, @@ -465,12 +518,33 @@ export class TeamService { async createTaskForHuman( namespace: string, teamId: string, - input: { title: string; assigneeSessionId?: string | null; dependsOn?: string[]; requirementId?: string } + input: { title: string; assigneeSessionId?: string | null; dependsOn?: string[]; requirementId?: string }, + /** Session caller (Lead via MCP). Absent = human web caller. */ + actorSessionId: string | null = null ): Promise { const team = this.store.getTeam(teamId, namespace) if (!team) { throw new TeamServiceError('not_found', 'Team not found') } + let actor: { fromKind: 'human' | 'session'; fromSessionId: string | null; fromRole: string } = { + fromKind: 'human', + fromSessionId: null, + fromRole: '人类' + } + if (actorSessionId) { + const membership = this.requireMembership(actorSessionId, namespace) + if (membership.team.id !== team.id) { + throw new TeamServiceError('forbidden', 'Session is not a member of this team') + } + if (membership.team.leadSessionId !== actorSessionId) { + throw new TeamServiceError('forbidden', 'Only the Lead can create tasks for team members') + } + actor = { + fromKind: 'session', + fromSessionId: actorSessionId, + fromRole: membership.member.role + } + } const members = this.store.listMembers(team.id) const assigneeSessionId = input.assigneeSessionId ?? null if (assigneeSessionId && !members.some((member) => member.sessionId === assigneeSessionId)) { @@ -495,13 +569,13 @@ export class TeamService { const assigneeRole = members.find((member) => member.sessionId === assigneeSessionId)?.role this.store.appendMessage({ teamId: team.id, - fromKind: 'human', - fromSessionId: null, + fromKind: actor.fromKind, + fromSessionId: actor.fromSessionId, toKind: assigneeSessionId ? 'task' : 'broadcast', toSessionId: assigneeSessionId, kind: 'task-assign', text: input.title, - meta: { taskId: task.id, fromRole: '人类', ...(assigneeRole ? { role: assigneeRole } : {}) } + meta: { taskId: task.id, fromRole: actor.fromRole, ...(assigneeRole ? { role: assigneeRole } : {}) } }) this.publishUpdate(team) @@ -510,9 +584,9 @@ export class TeamService { await this.runtime.deliverPeerMessage({ sessionId: assigneeSessionId, text: this.formatPeerText(team, members, { - fromKind: 'human', - fromSessionId: null, - fromRole: '人类', + fromKind: actor.fromKind, + fromSessionId: actor.fromSessionId, + fromRole: actor.fromRole, text: `你有一个新任务：${input.title}\n任务 id：${task.id}` }) }) @@ -630,6 +704,38 @@ export class TeamService { } } + // Reassignment is only safe before work starts: a fresh member has no + // access to the previous session's conversation or worktree. Anything + // in flight must be blocked and re-created with a fresh brief instead. + if (patch.assigneeSessionId !== undefined + && patch.assigneeSessionId !== task.assigneeSessionId + && task.status !== 'todo' + ) { + throw new TeamServiceError( + 'invalid', + 'Only todo tasks can be reassigned; mark the current task blocked and create a new one with a brief for the new member' + ) + } + + // One doing task per member: a second "doing" would silently split the + // member's attention and let the Lead over-assign without noticing. + const nextAssignee = patch.assigneeSessionId !== undefined + ? patch.assigneeSessionId + : task.assigneeSessionId + if (sessionId && patch.status === 'doing' && nextAssignee) { + const busy = this.store.listTasks(team.id).find((candidate) => + candidate.id !== taskId + && candidate.assigneeSessionId === nextAssignee + && candidate.status === 'doing' + ) + if (busy) { + throw new TeamServiceError( + 'invalid', + `该成员已有进行中的任务「${busy.title}」，请先完成或改派再开始新任务` + ) + } + } + const updated = this.store.updateTask(taskId, { ...(patch.status !== undefined ? { status: patch.status } : {}), ...(patch.assigneeSessionId !== undefined ? { assigneeSessionId: patch.assigneeSessionId } : {}), @@ -659,6 +765,30 @@ export class TeamService { }) this.publishUpdate(team) + // A todo task handed to another member: tell them, with the stored brief + // when there is one, so they are not starting from a bare title. + if (patch.assigneeSessionId !== undefined + && updated.assigneeSessionId + && updated.assigneeSessionId !== task.assigneeSessionId + && this.runtime + ) { + const meta = updated.meta ?? {} + const brief = typeof meta.brief === 'string' ? meta.brief.trim() : '' + try { + await this.runtime.deliverPeerMessage({ + sessionId: updated.assigneeSessionId, + text: this.formatPeerText(team, members, { + fromKind: membership ? 'session' : 'human', + fromSessionId: sessionId, + fromRole, + text: `任务「${updated.title}」已指派给你${brief ? `\n任务说明：${brief}` : ''}\n任务 id：${updated.id}` + }) + }) + } catch { + // Best-effort; the task board still shows the assignment. + } + } + // Done/blocked are lead-relevant: push those to the lead explicitly. if (membership && patch.status !== undefined && (patch.status === 'done' || patch.status === 'blocked') @@ -1243,6 +1373,29 @@ export class TeamService { // ------------------------------------------------------------------ helpers + /** + * Star topology: only the Lead opens a direct channel to another member. + * Members report with `to="lead"` and reach everyone else through the shared + * log (broadcast) or their tasks; peer DMs would wake another member behind + * the Lead's back and bypass the task board. + */ + private assertMemberTargetAllowed( + team: TeamRecord, + senderSessionId: string, + target: { toSessionId: string | null } + ): void { + if (!target.toSessionId) { + return + } + if (team.leadSessionId === senderSessionId || target.toSessionId === team.leadSessionId) { + return + } + throw new TeamServiceError( + 'forbidden', + 'Only the Lead can direct-message another member; report with to="lead" or broadcast to the team log instead' + ) + } + private requireMembership(sessionId: string, namespace: string): { team: TeamRecord; member: TeamMemberRecord } { const membership = this.store.findTeamBySession(sessionId, namespace) if (!membership) { @@ -1422,7 +1575,9 @@ function buildLeadBrief(team: TeamRecord): string { `你是 HAPI 团队「${team.name}」的 Lead。`, '', '职责：拆解任务、派生成员、汇总进展，必要时把决策升级给人类。', - '可用工具：team_status（成员/任务/需求/预算）、team_read（拉取团队消息，广播不会主动推送）、team_send（汇报/分派/通知人类）、team_task（任务列表/更新状态/交付物/依赖）、team_requirement（需求列表/状态/结论）、spawn_peer（派生成员，需用户批准）。', + '成员只向你汇报（成员之间不能直接发消息）；需要协作时用任务和依赖串联，不要让他们互相喊话。', + '可用工具：team_status（成员/任务/需求/预算）、team_read（拉取团队消息，广播不会主动推送）、team_send（汇报/分派/通知人类）、team_task（列表/建任务/更新状态/交付物/依赖）、team_requirement（需求列表/状态/结论）、spawn_peer（派生成员，需用户批准）。', + '复用 idle 成员时用 team_task action=create 给它建任务并指派，不要为了一个小任务新开成员；每个成员同一时间只能有一个 doing 任务。', '人类每条新消息会自动成为一个"需求"：派生任务时带上 requirementId（spawn_peer），完成后用 team_requirement 把需求标 done 并写一句 conclusion（结果）——人类只读结论即可。', '派生成员默认继承你的工具/模型/思考等级/权限，不需要手动指定；只有人类明确要求不同配置时才传覆盖参数。', '需要直接调用 hub API 时用团队级凭证 $HAPI_TEAM_TOKEN（只能访问本团队的消息/任务/状态），不要读取 ~/.hapi/settings.json 的凭证。', @@ -1449,7 +1604,7 @@ function buildAssignmentBrief( '', '团队规范：', '- 开工前可调用 team_read 拉取团队消息（广播不会主动推送）。', - '- 进展/完成/阻塞用 team_send 汇报（kind=status 或 task-update）。', + '- 进展/完成/阻塞用 team_send 汇报（kind=status 或 task-update），并且只发给 Lead（to="lead"）；成员之间不能直接发消息。', '- 任务状态用 team_task 更新：开工标 doing、卡住标 blocked、完成标 done 并附交付物（分支/文件/测试结果）；有依赖的任务需依赖先完成。', '- 汇报时带上 taskId（team_send 的 taskId 参数），消息会自动归入对应需求。', '- 需要 hub API 时用团队级凭证 $HAPI_TEAM_TOKEN 调 $HAPI_API_URL（只能访问本团队），不要读取 ~/.hapi/settings.json。', @@ -1482,6 +1637,19 @@ function isAcknowledgement(text: string): boolean { return /^(ok|okay|okey|收到|好|好的|好嘞|行|嗯+|谢谢|多谢|辛苦了|thx|thanks|👍)$/i.test(compact) } +/** Mirrors the web "待你确认" inbox: an unanswered decision from a member. */ +function isPendingHumanDecision(message: TeamMessageRecord): boolean { + if (message.fromKind !== 'session') { + return false + } + const meta = message.meta ?? {} + const isDecision = message.kind === 'decision' || meta.awaitingHuman === true + if (!isDecision) { + return false + } + return meta.humanRepliedAt === undefined && meta.humanDismissedAt === undefined +} + /** Human ask -> requirement title: first non-empty line, capped. */ function requirementTitle(text: string): string { const firstLine = text.split('\n').map((line) => line.trim()).find((line) => line.length > 0) ?? '' diff --git a/hub/src/teams/teamStore.ts b/hub/src/teams/teamStore.ts index 04539bf9..35f70ef0 100644 --- a/hub/src/teams/teamStore.ts +++ b/hub/src/teams/teamStore.ts @@ -478,6 +478,28 @@ export class TeamStore { return rows.map(mapMessageRow) } + /** Latest message timestamp per member session, for the team dashboard. */ + lastMessageAtBySession(teamId: string): Record { + const rows = this.db.prepare( + `SELECT from_session_id AS sessionId, MAX(created_at) AS at + FROM team_messages + WHERE team_id = ? AND from_session_id IS NOT NULL + GROUP BY from_session_id` + ).all(teamId) as Array<{ sessionId: string; at: number }> + const out: Record = {} + for (const row of rows) { + out[row.sessionId] = row.at + } + return out + } + + lastMessageAt(teamId: string): number | null { + const row = this.db.prepare( + 'SELECT MAX(created_at) AS at FROM team_messages WHERE team_id = ?' + ).get(teamId) as { at: number | null } | undefined + return row?.at ?? null + } + getMessage(teamId: string, seq: number): TeamMessageRecord | null { const row = this.db.prepare( 'SELECT * FROM team_messages WHERE team_id = ? AND seq = ?' diff --git a/hub/src/web/routes/teams.test.ts b/hub/src/web/routes/teams.test.ts index 2d4a022f..528181c7 100644 --- a/hub/src/web/routes/teams.test.ts +++ b/hub/src/web/routes/teams.test.ts @@ -11,7 +11,15 @@ import { TeamStore } from '../../teams/teamStore' import type { WebAppEnv } from '../middleware/auth' import { createTeamsRoutes } from './teams' -function createApp(getNamespace: () => string = () => 'alpha') { +function createApp( + getNamespace: () => string = () => 'alpha', + options: { + getMemberUsage?: ( + namespace: string, + sessionIds: string[] + ) => Map + } = {} +) { const events: SyncEvent[] = [] const sessions = new Map() sessions.set('sess-lead', { @@ -55,7 +63,7 @@ function createApp(getNamespace: () => string = () => 'alpha') { c.set('namespace', getNamespace()) await next() }) - app.route('/api', createTeamsRoutes(service)) + app.route('/api', createTeamsRoutes(service, options)) return { app, events, service, delivered, archived } } @@ -84,6 +92,64 @@ describe('Agent Team member routes', () => { expect(outsider.status).toBe(404) }) + it('returns a cross-team overview for the control room', async () => { + const usageCalls: string[][] = [] + const { app, service } = createApp(() => 'alpha', { + getMemberUsage: (_namespace, sessionIds) => { + usageCalls.push(sessionIds) + return new Map(sessionIds.map((sessionId) => [ + sessionId, + { last24hTokens: 100, totalTokens: 1_000 } + ])) + } + }) + const teamId = await createTeam(app) + const spawned = await app.request(`/api/teams/${teamId}/spawn`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ fromSessionId: 'sess-lead', role: 'builder', task: 'design' }) + }) + const spawnBody = await spawned.json() as { sessionId: string; taskId: string } + await service.updateTask(spawnBody.sessionId, 'alpha', spawnBody.taskId, { status: 'doing' }) + await service.sendMessage(spawnBody.sessionId, 'alpha', { + text: 'need a call', + to: 'human', + kind: 'decision' + }) + const human = await service.sendHumanMessage('alpha', teamId, { text: 'ship it' }) + await service.updateRequirement(null, 'alpha', human.meta?.requirementId as string, { + status: 'done' + }) + + const response = await app.request('/api/teams/overview') + expect(response.status).toBe(200) + const body = await response.json() as { + teams: Array<{ + id: string + pendingDecisions: number + requirements: { missingConclusion: number } + tasks: { doing: number } + members: Array<{ + sessionId: string + currentTask: { title: string } | null + lastReportAt: number | null + usage?: { last24hTokens: number; totalTokens: number } + }> + }> + } + expect(body.teams).toHaveLength(1) + const team = body.teams[0]! + expect(team.id).toBe(teamId) + expect(team.pendingDecisions).toBe(1) + expect(team.requirements.missingConclusion).toBe(1) + expect(team.tasks.doing).toBe(1) + const builder = team.members.find((member) => member.sessionId === spawnBody.sessionId) + expect(builder?.currentTask?.title).toBe('design') + expect(typeof builder?.lastReportAt).toBe('number') + expect(builder?.usage).toEqual({ last24hTokens: 100, totalTokens: 1_000 }) + expect(usageCalls[0]).toContain(spawnBody.sessionId) + }) + it('sends team messages with push for directed targets', async () => { const { app, delivered } = createApp() const teamId = await createTeam(app) diff --git a/hub/src/web/routes/teams.ts b/hub/src/web/routes/teams.ts index 1f047796..c818a52c 100644 --- a/hub/src/web/routes/teams.ts +++ b/hub/src/web/routes/teams.ts @@ -20,7 +20,16 @@ import type { WebAppEnv } from '../middleware/auth' * Agent Team API. Registered only when the hub has `teamsEnabled` — with the * feature off these paths do not exist (404), matching pre-feature behavior. */ -export function createTeamsRoutes(teams: TeamService): Hono { +export function createTeamsRoutes( + teams: TeamService, + options: { + /** Per-member token totals; omitted for non-owner namespaces. */ + getMemberUsage?: ( + namespace: string, + sessionIds: string[] + ) => Map + } = {} +): Hono { const app = new Hono() app.get('/teams', (c) => { @@ -37,6 +46,35 @@ export function createTeamsRoutes(teams: TeamService): Hono { return c.json({ enabled: true }) }) + // Cross-team control room. Registered before /teams/:id for the same reason. + app.get('/teams/overview', (c) => { + c.header('Cache-Control', 'no-store') + const namespace = c.get('namespace') + const overview = teams.getOverview(namespace) + const sessionIds = Array.from(new Set( + overview.flatMap((team) => team.members.map((member) => member.sessionId)) + )) + let usage: Map | undefined + if (sessionIds.length > 0 && options.getMemberUsage) { + try { + usage = options.getMemberUsage(namespace, sessionIds) + } catch { + // The dashboard still works without usage totals. + usage = undefined + } + } + return c.json({ + generatedAt: Date.now(), + teams: overview.map((team) => ({ + ...team, + members: team.members.map((member) => { + const memberUsage = usage?.get(member.sessionId) + return memberUsage ? { ...member, usage: memberUsage } : member + }) + })) + }) + }) + app.get('/teams/by-session/:sessionId', (c) => { c.header('Cache-Control', 'no-store') const sessionId = c.req.param('sessionId') @@ -166,7 +204,7 @@ export function createTeamsRoutes(teams: TeamService): Hono { assigneeSessionId: parsed.data.assigneeSessionId ?? null, dependsOn: parsed.data.dependsOn, requirementId: parsed.data.requirementId - }) + }, parsed.data.fromSessionId ?? null) c.header('Cache-Control', 'no-store') return c.json({ task }, 201) } catch (error) { diff --git a/hub/src/web/server.ts b/hub/src/web/server.ts index 828dc200..712ef99f 100644 --- a/hub/src/web/server.ts +++ b/hub/src/web/server.ts @@ -40,6 +40,7 @@ import { createClaudeModelsRoutes } from './routes/claudeModels' import { createPublicStudioRoutes, createStudioRoutes } from './routes/studios' import { createSessionShareRoutes } from './routes/sessionShares' import { createTeamsRoutes } from './routes/teams' +import { getUsageSummary } from '../sync/usageService' import type { SSEManager } from '../sse/sseManager' import type { VisibilityTracker } from '../visibility/visibilityTracker' import type { TeamService } from '../teams/teamService' @@ -323,9 +324,50 @@ function createWebApp(options: { app.route('/api', createMachinesRoutes(options.getSyncEngine, options.getTeamService, { dataDir: configuration.dataDir })) app.route('/api', createStorageRoutes(configuration.dbPath)) app.route('/api', createHubSettingsRoutes(configuration.dataDir, () => configuration.teamsEnabled)) + // Team dashboard usage: scanning the usage ledger is heavier than the other + // reads, so keep one namespace-wide per-session map for a minute and filter + // it per request. Usage stays owner-only, matching /api/usage/summary. + let memberUsageCache: { + at: number + namespace: string + map: Map + } | null = null + const getMemberUsage = ( + namespace: string, + sessionIds: string[] + ): Map => { + if (namespace !== 'default') { + return new Map() + } + const now = Date.now() + if (!memberUsageCache || memberUsageCache.namespace !== namespace || now - memberUsageCache.at > 60_000) { + const collect = (range: string) => new Map( + (getUsageSummary(options.store, namespace, range, 'UTC', { includeBySession: true }).bySession ?? []) + .map((bucket) => [bucket.key, bucket.totalTokens]) + ) + const recent = collect('1d') + const total = collect('all') + const map = new Map() + for (const sessionId of new Set([...recent.keys(), ...total.keys()])) { + map.set(sessionId, { + last24hTokens: recent.get(sessionId) ?? 0, + totalTokens: total.get(sessionId) ?? 0 + }) + } + memberUsageCache = { at: now, namespace, map } + } + const out = new Map() + for (const sessionId of sessionIds) { + const entry = memberUsageCache.map.get(sessionId) + if (entry) { + out.set(sessionId, entry) + } + } + return out + } const teamService = options.getTeamService?.() ?? null if (teamService) { - app.route('/api', createTeamsRoutes(teamService)) + app.route('/api', createTeamsRoutes(teamService, { getMemberUsage })) } app.route('/api', createUsageRoutes(options.store, options.refreshUsage)) app.route('/api', createActivityRoutes(options.store)) diff --git a/scripts/dsh-web-token-wrapper.mjs b/scripts/dsh-web-token-wrapper.mjs new file mode 100755 index 00000000..ce75014b --- /dev/null +++ b/scripts/dsh-web-token-wrapper.mjs @@ -0,0 +1,52 @@ +#!/usr/bin/env node + +import { chmodSync, mkdirSync, renameSync, unlinkSync, writeFileSync } from 'node:fs' +import { dirname } from 'node:path' +import { spawn } from 'node:child_process' + +const dshBin = process.env.DSH_BIN?.trim() +const tokenFile = process.env.DSH_TOKEN_FILE?.trim() +if (!dshBin || !tokenFile) { + console.error('dsh web wrapper requires DSH_BIN and DSH_TOKEN_FILE') + process.exit(64) +} + +mkdirSync(dirname(tokenFile), { recursive: true, mode: 0o700 }) +try { unlinkSync(tokenFile) } catch (error) { + if (error?.code !== 'ENOENT') throw error +} + +const child = spawn(process.env.DSH_NODE?.trim() || process.execPath, [dshBin, ...process.argv.slice(2)], { + env: process.env, + stdio: ['inherit', 'pipe', 'inherit'] +}) + +let pending = '' +let tokenSaved = false +const consume = (chunk) => { + const text = String(chunk) + process.stdout.write(text) + pending += text + const lines = pending.split(/\r?\n/u) + pending = lines.pop() ?? '' + for (const line of lines) { + const match = /dsh web:\s*https?:\/\/[^\s?]+(?:\?[^\s]*)?token=([A-Za-z0-9_-]+)/u.exec(line) + if (!match || tokenSaved) continue + const temporary = `${tokenFile}.tmp-${process.pid}` + writeFileSync(temporary, `${match[1]}\n`, { mode: 0o600 }) + chmodSync(temporary, 0o600) + renameSync(temporary, tokenFile) + tokenSaved = true + } +} +child.stdout.on('data', consume) +child.on('error', (error) => { + console.error(`dsh web wrapper failed to start: ${error.message}`) +}) +for (const signal of ['SIGTERM', 'SIGINT', 'SIGHUP']) { + process.on(signal, () => child.kill(signal)) +} +child.on('exit', (code, signal) => { + if (pending) consume('') + process.exit(signal ? 1 : (code ?? 1)) +}) diff --git a/shared/src/apiTypes.ts b/shared/src/apiTypes.ts index 1600e67a..42d200c9 100644 --- a/shared/src/apiTypes.ts +++ b/shared/src/apiTypes.ts @@ -563,6 +563,46 @@ export const CreateTeamRequestSchema = z.object({ export type CreateTeamRequest = z.infer +/** Cross-team dashboard read model: one screen for every team's live state. */ +export type TeamOverviewMemberStatus = 'idle' | 'working' | 'blocked' | 'offline' + +export type TeamOverviewMemberUsage = { + /** Tokens in the trailing 24 hours. */ + last24hTokens: number + /** Lifetime tokens for this member session. */ + totalTokens: number +} + +export type TeamOverviewMember = { + sessionId: string + role: string + status: TeamOverviewMemberStatus + /** The single task this member currently has in `doing`, if any. */ + currentTask: { id: string; title: string } | null + /** Last message this member posted into the team log. */ + lastReportAt: number | null + /** Present only for the hub owner namespace. */ + usage?: TeamOverviewMemberUsage +} + +export type TeamOverviewTeam = { + id: string + name: string + status: 'active' | 'archived' + leadSessionId: string | null + members: TeamOverviewMember[] + tasks: { total: number; todo: number; doing: number; blocked: number; done: number } + requirements: { total: number; open: number; missingConclusion: number } + /** Member messages with kind=decision still waiting for a human answer. */ + pendingDecisions: number + lastActivityAt: number | null +} + +export type TeamOverviewResponse = { + generatedAt: number + teams: TeamOverviewTeam[] +} + export const TeamSendMessageRequestSchema = z.object({ fromSessionId: z.string().min(1), text: z.string().min(1).max(20000), @@ -646,6 +686,8 @@ export const TeamTaskUpdateRequestSchema = z.object({ export type TeamTaskUpdateRequest = z.infer export const CreateTeamTaskRequestSchema = z.object({ + /** Session caller (CLI/MCP). Absent = human web caller. Only the Lead may pass this. */ + fromSessionId: z.string().min(1).optional(), title: z.string().min(1).max(200), assigneeSessionId: z.string().min(1).nullable().optional(), dependsOn: z.array(z.string().min(1)).max(20).optional(), @@ -1481,6 +1523,8 @@ export type UsageSummaryResponse = { daily: Array byAgent: UsageSummaryBucket[] byModel: UsageSummaryBucket[] + /** Per-session buckets; only populated when the caller asks for them. */ + bySession?: UsageSummaryBucket[] updatedAt: number } diff --git a/web/src/api/client.ts b/web/src/api/client.ts index f2f1ea65..dcb4f181 100644 --- a/web/src/api/client.ts +++ b/web/src/api/client.ts @@ -49,6 +49,7 @@ import type { SessionContentSearchResponse } from '@/types/api' import type { TeamDetail, TeamMessage, TeamSummary, TeamTask, TeamWithMembers } from '@/types/team' +import type { TeamOverviewResponse } from '@hapi/protocol/apiTypes' import type { CodexGoalRequest, CodexGoalResponse, @@ -570,6 +571,10 @@ export class ApiClient { return await this.request<{ teams: TeamWithMembers[] }>('/api/teams') } + async getTeamOverview(): Promise { + return await this.request('/api/teams/overview') + } + async getTeam(teamId: string): Promise { return await this.request(`/api/teams/${encodeURIComponent(teamId)}`) } diff --git a/web/src/components/teams/TeamOverviewPage.tsx b/web/src/components/teams/TeamOverviewPage.tsx new file mode 100644 index 00000000..6b030412 --- /dev/null +++ b/web/src/components/teams/TeamOverviewPage.tsx @@ -0,0 +1,242 @@ +import { useNavigate } from '@tanstack/react-router' +import type { TeamOverviewMember, TeamOverviewTeam } from '@hapi/protocol/apiTypes' +import { useAppContext } from '@/lib/app-context' +import { useTranslation } from '@/lib/use-translation' +import { useTeamOverview } from '@/hooks/queries/useTeamOverview' +import { memberStatusLabel, statusDotClass, type TranslateFn } from './teamStatus' + +function formatTokens(value: number): string { + if (value >= 1_000_000) return `${(value / 1_000_000).toFixed(1)}M` + if (value >= 1_000) return `${(value / 1_000).toFixed(1)}k` + return String(value) +} + +function formatRelative(t: TranslateFn, timestamp: number | null): string { + if (!timestamp) return t('dashboard.never') + const diff = Math.max(0, Date.now() - timestamp) + if (diff < 60_000) return t('dashboard.time.justNow') + const minutes = Math.floor(diff / 60_000) + if (minutes < 60) return t('dashboard.time.minutes', { n: minutes }) + const hours = Math.floor(minutes / 60) + if (hours < 24) return t('dashboard.time.hours', { n: hours }) + return t('dashboard.time.days', { n: Math.floor(hours / 24) }) +} + +function SummaryTile(props: { label: string; value: number | string; accent?: 'amber' | 'emerald' }) { + const accent = props.accent === 'amber' + ? 'text-amber-600' + : props.accent === 'emerald' + ? 'text-emerald-600' + : 'text-[var(--app-fg)]' + return ( +
+
{props.value}
+
{props.label}
+
+ ) +} + +function MemberRow(props: { + member: TeamOverviewMember + isLead: boolean + onOpen: () => void +}) { + const { t } = useTranslation() + const { member } = props + return ( + + ) +} + +function TeamCard(props: { + team: TeamOverviewTeam + onOpenTeam: () => void + onOpenSession: (sessionId: string) => void +}) { + const { t } = useTranslation() + const { team } = props + const active = team.status === 'active' + return ( +
+
+ + {!active ? ( + + {t('team.section.archived')} + + ) : null} + {team.pendingDecisions > 0 ? ( + + {t('dashboard.team.pending', { n: team.pendingDecisions })} + + ) : null} + {team.requirements.missingConclusion > 0 ? ( + + {t('dashboard.team.missing', { n: team.requirements.missingConclusion })} + + ) : null} + + {formatRelative(t, team.lastActivityAt)} + +
+ +
+ {t('dashboard.team.tasks', { + doing: team.tasks.doing, + todo: team.tasks.todo, + done: team.tasks.done, + blocked: team.tasks.blocked, + })} + {t('dashboard.team.requirements', { + open: team.requirements.open, + total: team.requirements.total, + })} +
+ +
+ {team.members.length === 0 ? ( +
{t('team.section.noMembers')}
+ ) : null} + {team.members.map((member) => ( + props.onOpenSession(member.sessionId)} + /> + ))} +
+
+ ) +} + +export function TeamOverviewPage() { + const { api } = useAppContext() + const { t } = useTranslation() + const navigate = useNavigate() + const { overview, isLoading, error, refetch } = useTeamOverview(api) + + const teams = overview?.teams ?? [] + const members = teams.flatMap((team) => team.members) + const working = members.filter((member) => member.status === 'working').length + const blocked = members.filter((member) => member.status === 'blocked').length + const pendingDecisions = teams.reduce((sum, team) => sum + team.pendingDecisions, 0) + const missingConclusions = teams.reduce((sum, team) => sum + team.requirements.missingConclusion, 0) + + return ( +
+
+
+ +
+

{t('dashboard.title')}

+

{t('dashboard.subtitle')}

+
+ +
+ + {error ? ( +
+ {t('dashboard.error')}: {error} +
+ ) : null} + + {isLoading && !overview ? ( +
{t('dashboard.loading')}
+ ) : teams.length === 0 ? ( +
+ {t('dashboard.empty')} +
+ ) : ( + <> +
+ + + 0 ? 'amber' : undefined} /> + 0 ? 'amber' : undefined} /> +
+ {blocked > 0 ? ( +
+ {t('dashboard.blockedMembers', { n: blocked })} +
+ ) : null} +
+ {teams.map((team) => ( + navigate({ + to: '/sessions/teams/$teamId', + params: { teamId: team.id }, + })} + onOpenSession={(sessionId) => navigate({ + to: '/sessions/$sessionId', + params: { sessionId }, + })} + /> + ))} +
+ + )} +
+
+ ) +} diff --git a/web/src/components/teams/TeamSidebarSection.tsx b/web/src/components/teams/TeamSidebarSection.tsx index 8b0e4ced..22c76c32 100644 --- a/web/src/components/teams/TeamSidebarSection.tsx +++ b/web/src/components/teams/TeamSidebarSection.tsx @@ -22,6 +22,8 @@ export function TeamSidebarSection(props: { sessions: SessionSummary[] selectedTeamId?: string | null selectedSessionId?: string | null + overviewSelected?: boolean + onSelectOverview: () => void onSelectTeam: (teamId: string) => void onSelectSession: (sessionId: string) => void }) { @@ -39,19 +41,34 @@ export function TeamSidebarSection(props: { }) }, [props.selectedTeamId]) - if (props.teams.length === 0) { - return null - } - const sessionsById = new Map(props.sessions.map((session) => [session.id, session])) return (
-
- {t('team.section.title')} -
-
- {props.teams.map((team) => { + + {props.teams.length === 0 ? null : ( + <> +
+ {t('team.section.title')} +
+
+ {props.teams.map((team) => { const isExpanded = expanded.has(team.id) return (
@@ -143,8 +160,10 @@ export function TeamSidebarSection(props: { ) : null}
) - })} -
+ })} +
+ + )}
) } diff --git a/web/src/hooks/queries/useTeamOverview.ts b/web/src/hooks/queries/useTeamOverview.ts new file mode 100644 index 00000000..17efc813 --- /dev/null +++ b/web/src/hooks/queries/useTeamOverview.ts @@ -0,0 +1,36 @@ +import { useQuery } from '@tanstack/react-query' +import type { TeamOverviewResponse } from '@hapi/protocol/apiTypes' +import type { ApiClient } from '@/api/client' +import { queryKeys } from '@/lib/query-keys' + +const OVERVIEW_STALE_TIME_MS = 5_000 + +export function useTeamOverview(api: ApiClient | null, enabled = true): { + overview: TeamOverviewResponse | null + isLoading: boolean + error: string | null + refetch: () => Promise +} { + const query = useQuery({ + queryKey: queryKeys.teamOverview, + queryFn: async () => { + if (!api) { + throw new Error('API unavailable') + } + return await api.getTeamOverview() + }, + enabled: Boolean(api) && enabled, + staleTime: OVERVIEW_STALE_TIME_MS, + // Status is derived from live session activity; a light poll keeps the + // control room honest while it is open. Usage is cached hub-side. + refetchInterval: 15_000, + retry: false, + }) + + return { + overview: query.data ?? null, + isLoading: query.isLoading, + error: query.error instanceof Error ? query.error.message : query.error ? 'Failed to load control room' : null, + refetch: query.refetch, + } +} diff --git a/web/src/hooks/useSSE.ts b/web/src/hooks/useSSE.ts index 6cf59f59..4e62af3c 100644 --- a/web/src/hooks/useSSE.ts +++ b/web/src/hooks/useSSE.ts @@ -444,6 +444,7 @@ export function useSSE(options: { } if (shouldInvalidateTeamsList) { tasks.push(queryClient.invalidateQueries({ queryKey: queryKeys.teams })) + tasks.push(queryClient.invalidateQueries({ queryKey: queryKeys.teamOverview })) } for (const teamId of teamIds) { tasks.push(queryClient.invalidateQueries({ queryKey: queryKeys.team(teamId) })) diff --git a/web/src/lib/locales/en.ts b/web/src/lib/locales/en.ts index 15106056..e01c3ba1 100644 --- a/web/src/lib/locales/en.ts +++ b/web/src/lib/locales/en.ts @@ -1514,6 +1514,30 @@ export default { 'session.notFoundReturningToShare': 'Session not found. Returning to the share page…', 'team.section.title': 'Teams', 'team.section.archived': 'archived', + 'dashboard.nav': 'Control room', + 'dashboard.title': 'Agent control room', + 'dashboard.subtitle': 'Every team, member status, current task and pending decisions.', + 'dashboard.refresh': 'Refresh', + 'dashboard.loading': 'Loading control room…', + 'dashboard.error': 'Failed to load the control room', + 'dashboard.empty': 'No teams yet. Create one from New Session → Team.', + 'dashboard.summary.teams': 'Teams', + 'dashboard.summary.working': 'Working now', + 'dashboard.summary.decisions': 'Waiting for you', + 'dashboard.summary.missing': 'Done without conclusion', + 'dashboard.blockedMembers': '{n} member(s) blocked', + 'dashboard.team.pending': '{n} decision(s)', + 'dashboard.team.missing': '{n} without conclusion', + 'dashboard.team.tasks': '{doing} doing · {todo} todo · {done} done · {blocked} blocked', + 'dashboard.team.requirements': 'Requirements {open}/{total} open', + 'dashboard.member.noTask': 'No task in progress', + 'dashboard.member.lastReport': 'Last report', + 'dashboard.member.usage': 'Tokens (24h / total)', + 'dashboard.never': 'never', + 'dashboard.time.justNow': 'just now', + 'dashboard.time.minutes': '{n}m ago', + 'dashboard.time.hours': '{n}h ago', + 'dashboard.time.days': '{n}d ago', 'team.title': 'Team', 'team.badge': 'Team', 'team.back': 'Back', diff --git a/web/src/lib/locales/zh-CN.ts b/web/src/lib/locales/zh-CN.ts index 42a96c5d..66c6e6f3 100644 --- a/web/src/lib/locales/zh-CN.ts +++ b/web/src/lib/locales/zh-CN.ts @@ -1513,6 +1513,30 @@ export default { 'session.notFoundReturningToShare': '会话不存在，即将返回分享页面…', 'team.section.title': '团队', 'team.section.archived': '已归档', + 'dashboard.nav': '总控台', + 'dashboard.title': 'Agent 总控台', + 'dashboard.subtitle': '所有团队、成员状态、当前任务与待你确认的事项。', + 'dashboard.refresh': '刷新', + 'dashboard.loading': '正在加载总控台…', + 'dashboard.error': '加载总控台失败', + 'dashboard.empty': '还没有团队。到「新建会话 → 团队」创建一个。', + 'dashboard.summary.teams': '团队', + 'dashboard.summary.working': '正在工作', + 'dashboard.summary.decisions': '待你确认', + 'dashboard.summary.missing': '完成但缺结论', + 'dashboard.blockedMembers': '{n} 个成员被阻塞', + 'dashboard.team.pending': '{n} 个待确认', + 'dashboard.team.missing': '{n} 个缺结论', + 'dashboard.team.tasks': '{doing} 进行中 · {todo} 待办 · {done} 完成 · {blocked} 阻塞', + 'dashboard.team.requirements': '需求 {open}/{total} 未完成', + 'dashboard.member.noTask': '当前没有进行中的任务', + 'dashboard.member.lastReport': '最后汇报', + 'dashboard.member.usage': 'Token（近24h / 累计）', + 'dashboard.never': '从未', + 'dashboard.time.justNow': '刚刚', + 'dashboard.time.minutes': '{n} 分钟前', + 'dashboard.time.hours': '{n} 小时前', + 'dashboard.time.days': '{n} 天前', 'team.title': '团队', 'team.badge': '团队', 'team.back': '返回', diff --git a/web/src/lib/query-keys.ts b/web/src/lib/query-keys.ts index 6c46753d..400dca20 100644 --- a/web/src/lib/query-keys.ts +++ b/web/src/lib/query-keys.ts @@ -50,5 +50,6 @@ export const queryKeys = { studio: (studioId: string) => ['studio', studioId] as const, teams: ['teams'] as const, team: (teamId: string) => ['team', teamId] as const, + teamOverview: ['team-overview'] as const, teamMessages: (teamId: string) => ['team-messages', teamId] as const, } diff --git a/web/src/router.tsx b/web/src/router.tsx index 7c68c4b6..995a8fd0 100644 --- a/web/src/router.tsx +++ b/web/src/router.tsx @@ -24,6 +24,7 @@ import { App } from '@/App' import { SessionChat } from '@/components/SessionChat' import { SessionList } from '@/components/SessionList' import { TeamChatPage } from '@/components/teams/TeamChatPage' +import { TeamOverviewPage } from '@/components/teams/TeamOverviewPage' import { TeamSidebarSection } from '@/components/teams/TeamSidebarSection' import { SessionShareManagerDialog } from '@/components/SessionShareManagerDialog' import { SidebarShowButton } from '@/components/SidebarToggle' @@ -260,7 +261,9 @@ function SessionsPage() { [machines] ) const sessionMatch = matchRoute({ to: '/sessions/$sessionId', fuzzy: true }) - const selectedSessionId = sessionMatch && sessionMatch.sessionId !== 'new' ? sessionMatch.sessionId : null + const selectedSessionId = sessionMatch && sessionMatch.sessionId !== 'new' && sessionMatch.sessionId !== 'dashboard' + ? sessionMatch.sessionId + : null const teamMatch = matchRoute({ to: '/sessions/teams/$teamId', fuzzy: true }) const selectedTeamId = teamMatch ? teamMatch.teamId : null // Team member sessions live inside their team in the sidebar, not as @@ -399,6 +402,11 @@ function SessionsPage() { sessions={sessions} selectedTeamId={selectedTeamId} selectedSessionId={selectedSessionId} + overviewSelected={pathname === '/sessions/dashboard'} + onSelectOverview={() => navigate({ + to: '/sessions/dashboard', + ...PRESERVE_SESSION_SIDEBAR_SCROLL, + })} onSelectTeam={(teamId) => navigate({ to: '/sessions/teams/$teamId', params: { teamId }, @@ -1524,6 +1532,12 @@ type NewSessionSearch = { teamName?: string } +const teamOverviewRoute = createRoute({ + getParentRoute: () => sessionsRoute, + path: 'dashboard', + component: TeamOverviewPage, +}) + const teamChatRoute = createRoute({ getParentRoute: () => sessionsRoute, path: 'teams/$teamId', @@ -1682,6 +1696,7 @@ export const routeTree = rootRoute.addChildren([ sessionsRoute.addChildren([ sessionsIndexRoute, newSessionRoute, + teamOverviewRoute, teamChatRoute, sessionDetailRoute.addChildren([ sessionTerminalRoute,