feat(teams,dsh): overview dashboard, lead task auth, and DSH launch tokens

- teams: cross-team overview read model + route/query/page; enforce member target visibility and restrict member task creation to the Lead - dsh: accept 0.2.x launch tokens from the printed URL, HAPI_DSH_TOKEN, or HAPI_DSH_TOKEN_FILE; bundled wrapper captures the printed token securely; per-session agent availability and remote event/auth handling - usage: support the 1d range and per-session totals in summaries - cli/web: team prompt/client plumbing, availability gating, locale and route wiring for the overview page Co-Authored-By: Claude Code <noreply@anthropic.com> Co-Authored-By: HAPI <noreply@hapi.run>
2026-11-14 02:51:43 +00:00 · 2026-11-05 02:22:50 +08:00
parent bc6b1577fa
commit 362b1fe47e
28 changed files with 1666 additions and 70 deletions
@@ -108,7 +108,21 @@ async function getDshWebAvailability(
 ): Promise<AgentAvailabilityEntry> {
    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' }
    }
@@ -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<any, any>('team_send', {
-            description: 'Agent Team: send a message to teammates. to="all" (default) writes to the shared log only; to="<role or session id prefix>" 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=<seq> 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=<seq> 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<any, any>('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 };
                }
@@ -393,7 +393,7 @@ export async function runHappyMcpStdioBridge(argv: string[]): Promise<void> {
      {
        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="<role or session id prefix>" 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=<seq> 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=<seq> 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<void> {
      {
        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)'),
        }),
      },
      {
@@ -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<string, unknown>; 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<string, unknown> : 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')
+    })
 })
@@ -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<string, unknown>

@@ -26,6 +29,7 @@ export type DshSessionSummary = {
    origin?: 'subagent'
    cwd?: string
    agentPreset?: string
+    agentAvailable?: boolean
    projections?: {
        asOfSeq: number
        values: Record<string, unknown>
@@ -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<void> {
+        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<void> {
+        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<void> {
+        // 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<void> {
+        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<void> {
+        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<number> {
+        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<void> {
+        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<string, string> }
+            ) => 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<string, string>): Record<string, string> {
+        return {
+            ...(this.authCookie ? { cookie: this.authCookie } : {}),
+            ...(extra ?? {})
+        }
+    }
+
+    private async ensureAuth(signal?: AbortSignal): Promise<boolean> {
+        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) {
@@ -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<DshQueueMode>((mode) => hashObject(mode))
    const pendingTurns = new Map<string, PendingTurn>()
@@ -297,6 +301,7 @@ export async function runDsh(opts: {
    let nativeTurnStateObserved = false
    let keepAliveInterval: ReturnType<typeof setInterval> | 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
            }
@@ -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<TeamTaskView> {
+    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 {
@@ -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.