09-订阅 DSH 事件与五种分发模式

ctx.on('order/submit', async (order, next) => {
    if (order.amount > 10000) return '被拦下'   // 不调 next()：后面的监听器和真正下单都不发生
    return next()                               // 放行：最里层才真正下单，返回值经这里往外传
})

/**
 * 我们自己声明的五个事件。
 *
 * 一个事件名只属于一种分发模式，所以五个模式各有自己的名字——这不是为了
 * 演示而拆的，harness 里的事件就是这么定的（`tools/result` 永远是广播，
 * `agent/pre-step` 永远是管道）。
 *
 * `declare module '@deepseek-ai/cordis'` 是类型层面的登记：Cordis 把所有
 * 插件声明的事件合并成一张总表，`ctx.emit` / `ctx.on` 的名字与参数才有类型。
 */
import type {} from '@deepseek-ai/cordis'

/** 一次实验带上的编号，以及监听器往里写话的轨迹数组。 */
export interface LabEvent {
    /** 本次实验的编号，探针每次请求自增。 */
    readonly run: number
    /** 令 `pipe` 的最外层监听器不调 `next()`，用来演示短路。 */
    readonly veto: boolean
    /** 监听器把"我跑了"写进这里；派发方读它，一次请求的结果就齐了。 */
    readonly trace: string[]
}

declare module '@deepseek-ai/cordis' {
    interface Events {
        /** 广播：每个监听器都跑一遍，返回值没人看。 */
        'event-lab/note'(payload: LabEvent): void
        /** 并发：每个监听器都跑一遍，派发方等最慢的那个。 */
        'event-lab/fanout'(payload: LabEvent): Promise<void>
        /** 顺序：依次 await，某个监听器返回非空值就停下。 */
        'event-lab/chain'(payload: LabEvent): string | void | Promise<string | void>
        /** 短路：同步依次调用，某个监听器返回非空值就停下。 */
        'event-lab/veto'(payload: LabEvent): string | void
        /** 管道：监听器包住 `next()`，不调 `next()` 就否决后面的监听器与内置行为。 */
        'event-lab/pipe'(payload: LabEvent, next: () => Promise<string>): Promise<string>
    }
}

/**
 * 演示用的五个监听器，一个分发模式一个。
 *
 * 这个插件不派发事件，只监听——派发方在 dsh-event-probe.ts 里。
 * 两边除了事件名（和它带来的 payload 类型）之外互不认识，跟 08-替换内置能力的提供方与消费者的
 * 「消费者只认接口」是同一回事。
 */
import { setTimeout as sleep } from 'node:timers/promises'
import type { Context } from '@deepseek-ai/cordis'
import type {} from './events.ts'

export const name = 'dsh-event-lab'

export function apply(ctx: Context): void {
    // 广播：两个监听器都会跑，顺序就是注册顺序，返回值被忽略。
    ctx.on('event-lab/note', (payload) => { payload.trace.push(`note-A(#${payload.run})`) })
    ctx.on('event-lab/note', (payload) => { payload.trace.push(`note-B(#${payload.run})`) })

    // 并发：两个异步监听器一起跑，谁先睡醒谁先写；派发方等最慢的那个。
    ctx.on('event-lab/fanout', async (payload) => {
        await sleep(120)
        payload.trace.push(`fanout-A(#${payload.run}，睡了 120ms)`)
    })
    ctx.on('event-lab/fanout', async (payload) => {
        await sleep(20)
        payload.trace.push(`fanout-B(#${payload.run}，睡了 20ms)`)
    })

    // 顺序：依次 await；B 返回了一个字符串，C 不再执行。
    ctx.on('event-lab/chain', async (payload) => { payload.trace.push(`chain-A(#${payload.run})`) })
    ctx.on('event-lab/chain', (payload) => {
        payload.trace.push(`chain-B(#${payload.run})`)
        return 'chain-B 叫停'
    })
    ctx.on('event-lab/chain', (payload) => { payload.trace.push(`chain-C(#${payload.run})`) })

    // 短路：同步版；A 返回了字符串，B 不再执行。
    ctx.on('event-lab/veto', (payload) => {
        payload.trace.push(`veto-A(#${payload.run})`)
        return 'veto-A 叫停'
    })
    ctx.on('event-lab/veto', (payload) => { payload.trace.push(`veto-B(#${payload.run})`) })

    // 管道：先注册的在最外层，最后兜底的是派发方传进来的那段"内置行为"。
    ctx.on('event-lab/pipe', async (payload, next) => {
        payload.trace.push(`pipe-A 进(#${payload.run})`)
        if (payload.veto) {
            payload.trace.push('pipe-A 没调 next，直接返回')
            return '被 pipe-A 否决'
        }
        const inner = await next()
        payload.trace.push('pipe-A 出')
        return `A(${inner})`
    })
    ctx.on('event-lab/pipe', async (payload, next) => {
        payload.trace.push('pipe-B')
        const builtin = await next()
        return `B(${builtin})`
    })

    // 07-配置与源码的热重载说过：ctx.logger 不往终端打，要肉眼可见的证据就自己 console.log。
    console.log(`[${name}] 5 个监听器已就位`)
}

/**
 * 一扇窗：用 curl 派发一次自己声明的事件，把监听器的轨迹原样回传。
 *
 * 五种分发模式各派发一个事件，好让"广播 / 并发 / 顺序 / 短路 / 管道"
 * 的差别在一次 curl 里看得见——真实插件不会为了看这个而发事件，这是
 * 演示与排查用的探针。08-替换内置能力的提供方与消费者的探针也是这个角色。
 */
import type { Context } from '@deepseek-ai/cordis'
import type { LabEvent } from './events.ts'
import type {} from './events.ts'
import type {} from '@deepseek-ai/dsh-host-webserver'

export const name = 'dsh-event-probe'

export const inject = ['webServer']

const PATH = '/dsh-event-lab/fire'

export function apply(ctx: Context): void {
    let run = 0
    ctx.effect(() => ctx.webServer.register({
        kind: 'exact',
        path: PATH,
        handler: async (req, res) => {
            const url = new URL(req.url ?? '/', 'http://127.0.0.1')
            const mode = url.searchParams.get('mode') ?? 'emit'
            const payload: LabEvent = {
                run: ++run,
                veto: url.searchParams.get('veto') === '1',
                trace: [],
            }
            const startedAt = Date.now()
            const returned = await fire(ctx, mode, payload)
            res.writeHead(200, { 'content-type': 'application/json; charset=utf-8' })
            res.end(JSON.stringify({
                mode,
                returned,
                elapsedMs: Date.now() - startedAt,
                trace: payload.trace,
            }, null, 2))
        },
    }))
}

/** 按模式派发。`waterfall` 多传一个参数：链条最内层的"内置行为"。 */
async function fire(ctx: Context, mode: string, payload: LabEvent): Promise<unknown> {
    switch (mode) {
        case 'emit':
            ctx.emit('event-lab/note', payload)
            return null
        case 'parallel':
            await ctx.parallel('event-lab/fanout', payload)
            return null
        case 'serial':
            return await ctx.serial('event-lab/chain', payload) ?? null
        case 'bail':
            return ctx.bail('event-lab/veto', payload) ?? null
        case 'waterfall':
            return await ctx.waterfall('event-lab/pipe', payload, async () => '内置行为')
        default:
            return `未知模式：${mode}`
    }
}

register(label: string, hooks: Hook[], callback: any, options: EventOptions): () => void {
    const method = options.prepend ? 'unshift' : 'push'
    return this.ctx.fiber.effect(() => {
        hooks[method]({ ctx: this.ctx, callback, ...options })
        return () => this.unregister(hooks, callback)
    }, label)
}

/**
 * 统计真实发生在 DSH 里的事件，并把计数挂到一条 HTTP 路由上。
 *
 * 它监听三件不同的事：
 *   - `tools/execute`（管道）：包住一次工具执行，量它花了多久；
 *   - `tools/result`（广播）：一次调用结束，记成功/失败；
 *   - `session/event`（广播）：每有一条会话记录落日志就触发一次。
 *
 * 三种用法各自对应一种分发模式的选择理由，正文里逐个说。
 */
import type { Context } from '@deepseek-ai/cordis'
import type {} from '@deepseek-ai/dsh-host-webserver'
import type {} from '@deepseek-ai/dsh-session'
import type {} from '@deepseek-ai/dsh-tools'
import type {} from '@deepseek-ai/dsh-user-approval'

export const name = 'dsh-event-stats'

// 故意不写模块级 inject：统计这件事本身不依赖任何服务，headless 里也要跑。
// 只有"把计数挂到路由上"这一段需要 webServer，所以用 04-注册 HTTP 路由与注入页面样式的写法单独等它。
const PATH = '/dsh-event-log/stats'

interface ToolStat {
    calls: number
    failures: number
    totalMs: number
}

export function apply(ctx: Context): void {
    const tools = new Map<string, ToolStat>()
    /** 会话记录（写进 JSONL 的那些）按类型计数。 */
    const recorded = new Map<string, number>()
    /** 审批结果按 outcome 计数。 */
    const approvals = new Map<string, number>()
    /** callId → 本次调用耗时，由上面的管道写、下面的广播取。 */
    const durations = new Map<string, number>()
    let sessions = 0

    // 管道：包住一次执行量时长。next() 的结果要原样返回（换个值就是替换
    // 这次调用的结果），也不能不调 next()——那等于不让这次调用真的执行。
    ctx.on('tools/execute', async (exec, next) => {
        const startedAt = Date.now()
        try {
            return await next()
        } finally {
            durations.set(exec.callId, Date.now() - startedAt)
        }
    })

    // 广播：此刻结果已经定型，监听器只观察、不改写。
    ctx.on('tools/result', (exec, result) => {
        const ms = durations.get(exec.callId)
        durations.delete(exec.callId)
        const stat = tools.get(exec.name) ?? { calls: 0, failures: 0, totalMs: 0 }
        stat.calls += 1
        if (result.isError) stat.failures += 1
        stat.totalMs += ms ?? 0
        tools.set(exec.name, stat)
        console.log(`[${name}] ${exec.name} ${result.isError ? '失败' : '成功'}${ms === undefined ? '' : ` ${ms}ms`}`)
    })

    /** 当前的计数快照：路由用它，一轮结束时的那行日志也用它。 */
    const snapshot = () => ({
        sessions,
        tools: Object.fromEntries([...tools].map(([toolName, stat]) => [toolName, {
            ...stat,
            avgMs: Math.round(stat.totalMs / stat.calls),
        }])),
        approvals: sorted(approvals),
        recorded: sorted(recorded),
    })

    // 广播：会话记录落一条触发一次。`event.type` 就是记录类型，`event.data` 是它的内容。
    ctx.on('session/event', (_session, event) => {
        bump(recorded, event.type)
        if (event.type === 'approval/decided') {
            bump(approvals, event.data.outcome)
        }
        // 一轮结束是个天然的结算点：headless 里没有路由可 curl，这行就是全部证据。
        if (event.type === 'turn/end') {
            console.log(`[${name}] 一轮结束：${JSON.stringify(snapshot())}`)
        }
    })

    ctx.on('session/created', () => { sessions += 1 })

    ctx.inject(['webServer'], (serverCtx) => {
        serverCtx.effect(() => serverCtx.webServer.register({
            kind: 'exact',
            path: PATH,
            handler: (_req, res) => {
                res.writeHead(200, { 'content-type': 'application/json; charset=utf-8' })
                res.end(JSON.stringify(snapshot(), null, 2))
            },
        }))
    })
}

function bump(counter: Map<string, number>, key: string): void {
    counter.set(key, (counter.get(key) ?? 0) + 1)
}

/** 次数多的在前；一样多就按名字。 */
function sorted(counter: Map<string, number>): Record<string, number> {
    return Object.fromEntries([...counter].sort((left, right) => right[1] - left[1] || left[0].localeCompare(right[0])))
}

async function 执行一次工具调用() {
    const result = await 派发waterfall('tools/execute')  // 管道监听器在这里面跑
    // ……归一化、post-execute、定稿……
    派发广播('tools/result')                              // 广播监听器在这里面跑
}