Add session bulk export and multi-target IM alarm delivery.

Ops can download every Host session as a ZIP for HQ analysis, fan key alarms to multiple WhatsApp targets, and the Plugins card copy stays short.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-09-06 14:11:54 +08:00
parent eb8299d155
commit 92279c1220
18 changed files with 3062 additions and 606 deletions

View file

@ -1,14 +1,25 @@
/**
* Optional IM sink for key alarms (soft-depends on dsh-im `ctx.dshIm`).
* Fan-out: one alarm can be delivered to every configured WhatsApp / IM target.
*/
import type { Context } from '@deepseek-ai/cordis'
import { formatAlarmPrompt, type KeyAlarmPayload } from './alarm-push.ts'
import {
normalizeImTarget,
resolveImTargets,
type ImDeliveryTarget,
} from './im-targets.ts'
export interface AlarmImDeliveryOptions {
enabled: boolean
botId: string
targetId: string
/** Preferred multi-target list (JSON settings or already parsed). */
targets?: readonly ImDeliveryTarget[]
/** Legacy single-target fields (used when `targets` is empty). */
botId?: string
targetId?: string
/** Raw `imTargets` settings string (optional; used with legacy fields). */
imTargets?: string
lang: string
}
@ -16,11 +27,32 @@ type DshIm = {
send: (botId: string, targetId: string, text: string, opts?: { signal?: AbortSignal }) => Promise<unknown>
}
function resolveTargets(options: AlarmImDeliveryOptions): ImDeliveryTarget[] {
if (options.targets && options.targets.length > 0) {
const out: ImDeliveryTarget[] = []
const seen = new Set<string>()
for (const entry of options.targets) {
const target = normalizeImTarget(entry)
if (!target) continue
const key = `${target.botId}::${target.targetId}`
if (seen.has(key)) continue
seen.add(key)
out.push(target)
}
return out
}
return resolveImTargets({
imTargets: options.imTargets,
imBotId: options.botId,
imTargetId: options.targetId,
})
}
/**
* Send a formatted key-alarm text through proactive IM delivery when configured.
* @param ctx - host cordis context (may lack `dshIm`).
* @param payload - matched alarm.
* @param options - settings snapshot for IM sink.
* @param options - settings snapshot for IM sink(s).
*/
export async function deliverAlarmToIm(
ctx: Context,
@ -28,11 +60,10 @@ export async function deliverAlarmToIm(
options: AlarmImDeliveryOptions,
): Promise<void> {
if (!options.enabled) return
const botId = options.botId.trim()
const targetId = options.targetId.trim()
if (!botId || !targetId) {
const targets = resolveTargets(options)
if (targets.length === 0) {
ctx.logger.warn(
'netxops alarm-im: enabled but imBotId/imTargetId empty — skip IM delivery (create a target in IM 投递设置 and paste botId+targetId)',
'netxops alarm-im: enabled but no delivery targets — skip IM delivery (pick one or more targets under Netx Ops → IM)',
)
return
}
@ -49,10 +80,24 @@ export async function deliverAlarmToIm(
}
const text = formatAlarmPrompt(payload, options.lang)
try {
await im.send(botId, targetId, text)
ctx.logger.info('netxops alarm-im: sent to botId=%s targetId=%s', botId, targetId)
} catch (error) {
ctx.logger.warn('netxops alarm-im: send failed: %s', error)
}
const results = await Promise.allSettled(
targets.map((target) => im.send(target.botId, target.targetId, text)),
)
results.forEach((result, index) => {
const target = targets[index]!
if (result.status === 'fulfilled') {
ctx.logger.info(
'netxops alarm-im: sent to botId=%s targetId=%s',
target.botId,
target.targetId,
)
return
}
ctx.logger.warn(
'netxops alarm-im: send failed botId=%s targetId=%s: %s',
target.botId,
target.targetId,
result.reason,
)
})
}

125
src/netx/im-targets.ts Normal file
View file

@ -0,0 +1,125 @@
/**
* Parse / format / resolve IM delivery target lists for multi-sink alarm push.
*/
export interface ImDeliveryTarget {
botId: string
targetId: string
}
/**
* Stable key for one bot+target pair.
* @param botId - opaque bot id.
* @param targetId - opaque target id.
*/
export function imTargetKey(botId: string, targetId: string): string {
return `${botId}::${targetId}`
}
/**
* Parse {@link imTargetKey}.
* @param value - encoded key.
*/
export function parseImTargetKey(value: string): ImDeliveryTarget {
const at = value.indexOf('::')
if (at <= 0) return { botId: '', targetId: '' }
return { botId: value.slice(0, at), targetId: value.slice(at + 2) }
}
/**
* Normalize one target; empty strings become absent.
* @param input - raw target.
*/
export function normalizeImTarget(input: unknown): ImDeliveryTarget | null {
if (input === null || typeof input !== 'object' || Array.isArray(input)) return null
const row = input as Record<string, unknown>
const botId = typeof row.botId === 'string' ? row.botId.trim() : ''
const targetId = typeof row.targetId === 'string' ? row.targetId.trim() : ''
if (!botId || !targetId) return null
return { botId, targetId }
}
/**
* Parse the settings JSON string for `imTargets`.
* @param text - stored settings value.
*/
export function parseImTargetsJson(text: string): ImDeliveryTarget[] {
const trimmed = text.trim()
if (!trimmed) return []
let parsed: unknown
try {
parsed = JSON.parse(trimmed)
} catch {
return []
}
if (!Array.isArray(parsed)) return []
const out: ImDeliveryTarget[] = []
const seen = new Set<string>()
for (const entry of parsed) {
const target = normalizeImTarget(entry)
if (!target) continue
const key = imTargetKey(target.botId, target.targetId)
if (seen.has(key)) continue
seen.add(key)
out.push(target)
}
return out
}
/**
* Serialize targets for the `imTargets` settings field.
* @param targets - selected sinks.
*/
export function formatImTargetsJson(targets: readonly ImDeliveryTarget[]): string {
const seen = new Set<string>()
const rows: ImDeliveryTarget[] = []
for (const entry of targets) {
const target = normalizeImTarget(entry)
if (!target) continue
const key = imTargetKey(target.botId, target.targetId)
if (seen.has(key)) continue
seen.add(key)
rows.push(target)
}
return rows.length === 0 ? '' : JSON.stringify(rows)
}
/**
* Resolve effective sinks: prefer `imTargets` JSON; else legacy single bot/target pair.
* @param settings - settings snapshot fields.
*/
export function resolveImTargets(settings: {
imTargets?: string
imBotId?: string
imTargetId?: string
}): ImDeliveryTarget[] {
const fromList = parseImTargetsJson(
typeof settings.imTargets === 'string' ? settings.imTargets : '',
)
if (fromList.length > 0) return fromList
const legacy = normalizeImTarget({
botId: settings.imBotId,
targetId: settings.imTargetId,
})
return legacy ? [legacy] : []
}
/**
* Toggle one catalog option in the selected set.
* @param current - current selection.
* @param target - option to toggle.
* @param selected - whether it should be selected after the toggle.
*/
export function setImTargetSelected(
current: readonly ImDeliveryTarget[],
target: ImDeliveryTarget,
selected: boolean,
): ImDeliveryTarget[] {
const key = imTargetKey(target.botId, target.targetId)
const without = current.filter(
(row) => imTargetKey(row.botId, row.targetId) !== key,
)
if (!selected) return without
const normalized = normalizeImTarget(target)
return normalized ? [...without, normalized] : without
}

479
src/netx/session-export.ts Normal file
View file

@ -0,0 +1,479 @@
/**
* Bulk session-log export for Ops: pack every durable session on this Host
* into one ZIP for offline HQ analysis. Soft-depends on sessionPersistence
* (JSONL raw artifacts); live sessions are flushed when a SessionStore is present.
*/
import { mkdirSync, writeFileSync } from 'node:fs'
import { homedir, hostname as osHostname } from 'node:os'
import { join } from 'node:path'
import { strToU8, zipSync, Zip, ZipDeflate } from 'fflate'
import type { Context } from '@deepseek-ai/cordis'
import {
NETXOPS_SESSIONS_EXPORT_PATH,
type SessionsExportStatus,
} from '../session-export-shared.ts'
export { NETXOPS_SESSIONS_EXPORT_PATH }
export type { SessionsExportStatus }
/** Default DEFLATE level for each ZIP entry. */
const COMPRESSION_LEVEL = 6
/** How many code units of artifact text one zip push carries. */
const PUSH_CHUNK_CODE_UNITS = 1 << 16
/** Subdir under `$DSH_HOME` where Ops bulk exports land. */
export const SESSIONS_EXPORT_DIRNAME = 'exports'
/** Duck-typed SessionPersistence surface used by export (no hard dep on dsh-session). */
export interface SessionPersistenceExportApi {
readonly supportsRawArtifacts: boolean
list: (signal?: AbortSignal) => Promise<readonly SessionExportHeader[]>
readRaw: (
id: string,
signal?: AbortSignal,
) => Promise<SessionExportRawArtifact | undefined>
}
/** Minimal durable session header fields copied into the export manifest. */
export interface SessionExportHeader {
readonly id: string
readonly version: number
readonly createdAt: number
readonly cwd?: string
readonly parentSession?: string
readonly origin?: string
readonly agentPreset?: string
readonly delegationDepth?: number
}
/** One verbatim persistence artifact. */
export interface SessionExportRawArtifact {
readonly meta: SessionExportHeader
readonly filename: string
readonly content: string
}
/** Optional live SessionStore flush barrier. */
interface SessionStoreFlushApi {
get?: (id: string) => unknown
flush?: (session: unknown) => Promise<void>
}
/** One exported file inside the bulk archive. */
type ZipTextEntry = { readonly path: string; readonly content: string }
/** Result of writing the bulk archive under `$DSH_HOME/exports`. */
export interface SessionsExportDiskResult {
readonly path: string
readonly filename: string
readonly exportDir: string
readonly sessionCountListed: number
readonly sessionCountIncluded: number
readonly sessionCountSkipped: number
readonly bytes: number
}
/** Harness home used for sessions + exports. */
export function resolveDshHome(): string {
const fromEnv = process.env.DSH_HOME?.trim()
if (fromEnv !== undefined && fromEnv.length > 0) return fromEnv
return join(homedir(), '.dsh')
}
/**
* Resolve sessionPersistence from the host context when the web profile mounted it.
* @param ctx - cordis host context.
*/
export function resolveSessionPersistence(ctx: Context): SessionPersistenceExportApi | undefined {
const fromGet = typeof (ctx as { get?: (name: string) => unknown }).get === 'function'
? (ctx as { get: (name: string) => unknown }).get('sessionPersistence')
: undefined
const persistence = (fromGet
?? (ctx as { sessionPersistence?: SessionPersistenceExportApi }).sessionPersistence) as
| SessionPersistenceExportApi
| undefined
if (!persistence || typeof persistence.list !== 'function' || typeof persistence.readRaw !== 'function') {
return undefined
}
return persistence
}
/**
* Resolve the optional live session store for flush-before-read.
* @param ctx - cordis host context.
*/
function resolveSessionStore(ctx: Context): SessionStoreFlushApi | undefined {
const fromGet = typeof (ctx as { get?: (name: string) => unknown }).get === 'function'
? (ctx as { get: (name: string) => unknown }).get('sessions')
: undefined
const sessions = (fromGet ?? (ctx as { sessions?: SessionStoreFlushApi }).sessions) as
| SessionStoreFlushApi
| undefined
if (!sessions || typeof sessions.get !== 'function' || typeof sessions.flush !== 'function') {
return undefined
}
return sessions
}
/**
* Preflight: whether bulk export can run and how many sessions would ship.
* @param ctx - cordis host context.
* @param signal - optional cancellation.
*/
export async function getSessionsExportStatus(
ctx: Context,
signal?: AbortSignal,
): Promise<SessionsExportStatus> {
const persistence = resolveSessionPersistence(ctx)
if (!persistence) {
return {
available: false,
sessionCount: 0,
supportsRawArtifacts: false,
reason: 'sessionPersistence unavailable — mount a JSONL session backend (web profile default)',
}
}
if (!persistence.supportsRawArtifacts) {
return {
available: false,
sessionCount: 0,
supportsRawArtifacts: false,
reason: 'persistence backend does not expose per-session raw artifacts (SQLite export unsupported)',
}
}
try {
const headers = await persistence.list(signal)
signal?.throwIfAborted()
return {
available: true,
sessionCount: headers.length,
supportsRawArtifacts: true,
}
} catch (error) {
return {
available: false,
sessionCount: 0,
supportsRawArtifacts: true,
reason: error instanceof Error ? error.message : String(error),
}
}
}
/**
* Sanitize one path segment for ZIP entry names.
* @param id - raw session id or hostname fragment.
*/
export function safePathSegment(id: string): string {
return id.replace(/[^A-Za-z0-9_-]/g, '_')
}
/**
* Browser download filename for a bulk export.
* @param exportedAt - Date used for the timestamp stamp.
* @param host - optional host label (defaults to os.hostname()).
*/
export function sessionsExportZipFilename(exportedAt: Date = new Date(), host = osHostname()): string {
const stamp = [
exportedAt.getUTCFullYear(),
String(exportedAt.getUTCMonth() + 1).padStart(2, '0'),
String(exportedAt.getUTCDate()).padStart(2, '0'),
'-',
String(exportedAt.getUTCHours()).padStart(2, '0'),
String(exportedAt.getUTCMinutes()).padStart(2, '0'),
String(exportedAt.getUTCSeconds()).padStart(2, '0'),
].join('')
return `dsh-sessions-${safePathSegment(host)}-${stamp}.zip`
}
async function flushLiveSession(
sessions: SessionStoreFlushApi | undefined,
id: string,
signal?: AbortSignal,
): Promise<void> {
signal?.throwIfAborted()
if (!sessions?.get || !sessions.flush) return
const live = sessions.get(id)
if (live === undefined || live === null) return
await sessions.flush(live)
signal?.throwIfAborted()
}
/**
* Yield ZIP text entries: manifest first, then each session artifact under sessions/<id>/.
* Missing artifacts are recorded in the manifest and skipped (fail-soft for Ops pickup).
* @param ctx - cordis host context.
* @param signal - cancellation.
*/
export async function* sessionsExportEntries(
ctx: Context,
signal?: AbortSignal,
): AsyncGenerator<ZipTextEntry> {
const persistence = resolveSessionPersistence(ctx)
if (!persistence) {
throw new Error('sessionPersistence unavailable')
}
if (!persistence.supportsRawArtifacts) {
throw new Error('persistence backend does not expose per-session raw artifacts')
}
const sessions = resolveSessionStore(ctx)
const headers = await persistence.list(signal)
signal?.throwIfAborted()
const exportedAt = new Date().toISOString()
const host = osHostname()
const included: Array<{
id: string
path: string
createdAt: number
cwd?: string
agentPreset?: string
parentSession?: string
origin?: string
}> = []
const skipped: Array<{ id: string; reason: string }> = []
const artifactEntries: ZipTextEntry[] = []
for (const header of headers) {
signal?.throwIfAborted()
const id = String(header.id)
try {
await flushLiveSession(sessions, id, signal)
const raw = await persistence.readRaw(id, signal)
signal?.throwIfAborted()
if (raw === undefined) {
skipped.push({ id, reason: 'no stored artifact' })
continue
}
const filename = raw.filename && raw.filename.length > 0 ? raw.filename : 'session.jsonl'
const path = `sessions/${safePathSegment(id)}/${filename}`
artifactEntries.push({ path, content: raw.content })
included.push({
id,
path,
createdAt: header.createdAt,
cwd: header.cwd,
agentPreset: header.agentPreset,
parentSession: header.parentSession !== undefined ? String(header.parentSession) : undefined,
origin: header.origin,
})
} catch (error) {
skipped.push({
id,
reason: error instanceof Error ? error.message : String(error),
})
}
}
const manifest = {
kind: 'dsh-netxops-sessions-export',
version: 1,
exportedAt,
hostname: host,
sessionCountListed: headers.length,
sessionCountIncluded: included.length,
sessionCountSkipped: skipped.length,
sessions: included,
skipped,
}
yield {
path: 'manifest.json',
content: `${JSON.stringify(manifest, null, 2)}\n`,
}
for (const entry of artifactEntries) {
signal?.throwIfAborted()
yield entry
}
}
/**
* Push artifact text into a deflate stream in bounded chunks without splitting
* surrogate pairs (lone high surrogates would corrupt UTF-8 as U+FFFD).
*/
async function pushArtifactChunks(
deflate: ZipDeflate,
content: string,
signal: AbortSignal,
): Promise<void> {
const encoder = new TextEncoder()
let offset = 0
let finalChunk: boolean
do {
signal.throwIfAborted()
let end = Math.min(offset + PUSH_CHUNK_CODE_UNITS, content.length)
if (end < content.length && end - offset > 1) {
const last = content.charCodeAt(end - 1)
if (last >= 0xd800 && last <= 0xdbff) end -= 1
}
finalChunk = end >= content.length
deflate.push(encoder.encode(content.slice(offset, end)), finalChunk)
offset = end
} while (!finalChunk)
}
/**
* Stream the bulk sessions ZIP as a WHATWG ReadableStream.
* @param ctx - cordis host context.
* @param signal - request / consumer cancellation.
*/
export function streamSessionsExportZip(
ctx: Context,
signal: AbortSignal,
): ReadableStream<Uint8Array> {
const consumerAbort = new AbortController()
const producerSignal = AbortSignal.any([signal, consumerAbort.signal])
let zip: Zip | undefined
let zipTerminated = false
const terminateZip = (): void => {
if (zip === undefined || zipTerminated) return
zipTerminated = true
zip.terminate()
}
return new ReadableStream<Uint8Array>({
start(controller) {
const archive = new Zip((error, data, final) => {
if (error) {
controller.error(error)
return
}
if (data.byteLength > 0) controller.enqueue(data)
if (final) controller.close()
})
zip = archive
void (async () => {
try {
for await (const entry of sessionsExportEntries(ctx, producerSignal)) {
const deflate = new ZipDeflate(entry.path, { level: COMPRESSION_LEVEL })
archive.add(deflate)
await pushArtifactChunks(deflate, entry.content, producerSignal)
}
archive.end()
} catch (error) {
terminateZip()
controller.error(error instanceof Error ? error : new Error(String(error)))
}
})()
},
cancel(reason) {
consumerAbort.abort(
reason instanceof Error ? reason : new Error('sessions export stream cancelled'),
)
terminateZip()
},
})
}
/**
* Write every durable session into `$DSH_HOME/exports/<filename>.zip`.
* This is the Ops-primary path: the absolute file location is returned so
* operators can copy the archive off the Host without relying on the browser
* download manager (which often swallows programmatic `<a download>` clicks).
* @param ctx - cordis host context.
* @param signal - optional cancellation.
*/
export async function writeSessionsExportZip(
ctx: Context,
signal?: AbortSignal,
): Promise<SessionsExportDiskResult> {
const status = await getSessionsExportStatus(ctx, signal)
if (!status.available) {
throw new Error(status.reason ?? 'sessions export unavailable')
}
const files: Record<string, Uint8Array> = {}
let included = 0
let skipped = 0
let listed = 0
for await (const entry of sessionsExportEntries(ctx, signal)) {
files[entry.path] = strToU8(entry.content)
if (entry.path === 'manifest.json') {
try {
const manifest = JSON.parse(entry.content) as {
sessionCountListed?: number
sessionCountIncluded?: number
sessionCountSkipped?: number
}
listed = typeof manifest.sessionCountListed === 'number' ? manifest.sessionCountListed : 0
included = typeof manifest.sessionCountIncluded === 'number' ? manifest.sessionCountIncluded : 0
skipped = typeof manifest.sessionCountSkipped === 'number' ? manifest.sessionCountSkipped : 0
} catch {
// Manifest parse is best-effort; ZIP still ships.
}
}
}
const zipped = zipSync(files, { level: COMPRESSION_LEVEL })
const exportDir = join(resolveDshHome(), SESSIONS_EXPORT_DIRNAME)
mkdirSync(exportDir, { recursive: true })
const filename = sessionsExportZipFilename()
const path = join(exportDir, filename)
writeFileSync(path, zipped)
return {
path,
filename,
exportDir,
sessionCountListed: listed || status.sessionCount,
sessionCountIncluded: included,
sessionCountSkipped: skipped,
bytes: zipped.byteLength,
}
}
/**
* Build the HTTP Response for GET /api/netxops.sessions.export.
* HEAD should use {@link sessionsExportHeadResponse} so listing does not start ZIP work.
* @param ctx - cordis host context.
* @param request - incoming fetch request.
*/
export async function sessionsExportResponse(
ctx: Context,
request: Request,
): Promise<Response> {
const status = await getSessionsExportStatus(ctx, request.signal)
if (!status.available) {
return new Response(status.reason ?? 'sessions export unavailable', {
status: status.supportsRawArtifacts === false && status.reason?.includes('raw artifacts')
? 501
: 500,
})
}
const filename = sessionsExportZipFilename()
const body = streamSessionsExportZip(ctx, request.signal)
return new Response(body, {
headers: {
'content-type': 'application/zip',
'content-disposition': `attachment; filename="${filename}"`,
'x-netxops-session-count': String(status.sessionCount),
},
})
}
/**
* Build the HTTP Response for HEAD /api/netxops.sessions.export (preflight only).
* @param ctx - cordis host context.
* @param request - incoming fetch request.
*/
export async function sessionsExportHeadResponse(
ctx: Context,
request: Request,
): Promise<Response> {
const status = await getSessionsExportStatus(ctx, request.signal)
if (!status.available) {
return new Response(null, {
status: status.supportsRawArtifacts === false && status.reason?.includes('raw artifacts')
? 501
: 500,
headers: {
'x-netxops-export-error': status.reason ?? 'sessions export unavailable',
},
})
}
const filename = sessionsExportZipFilename()
return new Response(null, {
status: 200,
headers: {
'content-type': 'application/zip',
'content-disposition': `attachment; filename="${filename}"`,
'x-netxops-session-count': String(status.sessionCount),
},
})
}