Mirror scheduled-run summaries into the origin effective session.

Pin origin at create time, keep isolated fire sessions for audit, and after IM/DSH delivery resume or open the origin chat to append the same summary for follow-ups.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-09-06 12:45:37 +08:00
parent 2e690c30c8
commit 29d946f3dd
12 changed files with 527 additions and 26 deletions

View file

@ -302,7 +302,7 @@ body>.dsh-ct-main{position:fixed;top:0;right:0;bottom:0;left:var(--dsh-ct-sideba
saveFailed: '本部署没有接受这些值,已保留供你修改。',
enabled: '启用调度器', timezone: '默认时区', historyLimit: '历史保留条数',
overlapPolicy: '重叠策略', misfirePolicy: '漏跑策略', policySkip: '跳过',
hint: '任务在左侧「新会话」下方进入。运行结果会出现在历史行里,点击记录会打开和普通会话一样的对话页,可以继续聊。',
hint: '每次执行会新开会话跑任务;摘要会镜像回创建时的有效会话(WhatsApp/Web),便于继续追问。完整工具轨迹仍可从历史打开。',
entry: '定时任务', entryLabel: '打开定时任务',
backToWorkspace: '返回工作区', backLabel: '返回工作区',
newJob: '新建任务', name: '名称', prompt: '提示词', kind: '日程',
@ -322,7 +322,7 @@ body>.dsh-ct-main{position:fixed;top:0;right:0;bottom:0;left:var(--dsh-ct-sideba
agentPreset: 'Agent Preset',
agentPresetDefault: '每次运行用当时的 Host 默认',
agentPresetHint: '指定后每次触发都挂载该 Preset;留空则跟随 Host 默认(创建时若从 WhatsApp 继承会自动写入)。',
editorLead: '保存后按日程在新会话里执行;若投递为 IM,结束后还会发摘要到 WhatsApp/IM。点击下方某次运行,右侧会打开那次对话,可以继续。',
editorLead: '到点仍会新开会话执行。结束后摘要会镜像回创建时的有效会话;IM 投递还会经 WhatsApp/IM 发出。日常追问请用原会话;完整工具轨迹可从下方运行记录打开。',
lastOutput: '上次输出', noOutput: '还没有输出。先立即运行一次。',
delivery: '投递',
deliveryDsh: 'DSH 侧栏会话',
@ -343,7 +343,7 @@ body>.dsh-ct-main{position:fixed;top:0;right:0;bottom:0;left:var(--dsh-ct-sideba
saveFailed: 'The deployment did not accept these values; they were left for you to correct.',
enabled: 'Enable scheduler', timezone: 'Default time zone', historyLimit: 'History retention',
overlapPolicy: 'Overlap policy', misfirePolicy: 'Misfire policy', policySkip: 'Skip',
hint: 'Open from under New Session. Run output appears in history; click a run to continue in the normal chat view.',
hint: 'Each fire runs in a fresh session; the summary is mirrored into the origin WhatsApp/Web session so you can follow up there. Full tool traces stay in run history.',
entry: 'Scheduled tasks', entryLabel: 'Open scheduled tasks',
backToWorkspace: 'Back to workspace', backLabel: 'Back to workspace',
newJob: 'New job', name: 'Name', prompt: 'Prompt', kind: 'Schedule',
@ -363,7 +363,7 @@ body>.dsh-ct-main{position:fixed;top:0;right:0;bottom:0;left:var(--dsh-ct-sideba
agentPreset: 'Agent preset',
agentPresetDefault: 'Use the Host default at fire time',
agentPresetHint: 'Pin a preset for every fire. Leave empty to follow the Host default (WhatsApp-created jobs inherit the chat preset automatically).',
editorLead: 'Runs start a new session. IM delivery also sends a summary via dshIm after the turn. Click a run below to open that conversation.',
editorLead: 'Fires still run in a new session. Afterward the summary is mirrored into the origin session; IM delivery also sends via WhatsApp/IM. Follow up in the origin chat; open a run below for the full tool trace.',
lastOutput: 'Last output', noOutput: 'No output yet. Run it once.',
delivery: 'Delivery',
deliveryDsh: 'DSH sidebar session',
@ -1230,6 +1230,16 @@ body>.dsh-ct-main{position:fixed;top:0;right:0;bottom:0;left:var(--dsh-ct-sideba
if (opened === false) setError('打不开这次对话,会话可能已被删除')
}
function currentSessionId(faces) {
try {
const snap = faces?.sessions?.list?.getSnapshot?.()
const current = snap?.current
return typeof current === 'string' && current.trim() ? current.trim() : ''
} catch {
return ''
}
}
async function save() {
try {
const atValue = form.kind === 'at' ? localInputToAt(form.at, form.timezone) : form.at
@ -1248,6 +1258,8 @@ body>.dsh-ct-main{position:fixed;top:0;right:0;bottom:0;left:var(--dsh-ct-sideba
if (selection.type === 'job' && selection.jobId) {
await api(`/jobs/${selection.jobId}`, { method: 'PATCH', body: JSON.stringify(payload) })
} else {
const originSessionId = currentSessionId(faces)
if (originSessionId) payload.origin = { kind: 'web', sessionId: originSessionId }
const created = await api('/jobs', { method: 'POST', body: JSON.stringify(payload) })
setSelection({ type: 'job', jobId: created.job.id })
}

View file

@ -266,6 +266,91 @@ export function callerSessionId(exec) {
).trim()
}
/**
* Normalize a durable origin binding for result mirroring.
* @param {object|null|undefined} input
* @returns {{ kind: 'web' | 'im', sessionId?: string, peer?: object } | null}
*/
export function normalizeOrigin(input) {
if (!input || typeof input !== 'object') return null
const sessionId = typeof input.sessionId === 'string' ? input.sessionId.trim() : ''
const peerSrc = input.peer && typeof input.peer === 'object' ? input.peer : null
const peerBotId = typeof peerSrc?.botId === 'string' ? peerSrc.botId.trim() : ''
const conversationKey = typeof peerSrc?.conversationKey === 'string'
? peerSrc.conversationKey.trim()
: ''
const kindRaw = String(input.kind || '').trim().toLowerCase()
const kind = kindRaw === 'im' || peerBotId
? 'im'
: (kindRaw === 'web' || sessionId ? 'web' : '')
if (!kind) return null
if (!sessionId && !(kind === 'im' && peerBotId && conversationKey)) return null
const origin = { kind }
if (sessionId) origin.sessionId = sessionId
if (kind === 'im' && peerBotId && conversationKey) {
origin.peer = {
botId: peerBotId,
conversationKey,
...typeof peerSrc.conversationId === 'string' && peerSrc.conversationId.trim()
? { conversationId: peerSrc.conversationId.trim() }
: {},
...peerSrc.kind === 'group' || peerSrc.kind === 'direct'
? { kind: peerSrc.kind }
: {},
}
}
return origin
}
function slimPeer(peer) {
if (!peer || typeof peer !== 'object') return null
const botId = typeof peer.botId === 'string' ? peer.botId.trim() : ''
const conversationKey = typeof peer.conversationKey === 'string'
? peer.conversationKey.trim()
: ''
if (!botId || !conversationKey) return null
return {
botId,
conversationKey,
...typeof peer.conversationId === 'string' && peer.conversationId.trim()
? { conversationId: peer.conversationId.trim() }
: {},
...peer.kind === 'group' || peer.kind === 'direct' ? { kind: peer.kind } : {},
}
}
/**
* Resolve origin for cron_create / sidebar create.
* Prefer an explicit origin; else capture the caller session (+ IM peer when present).
* @param {object} args
* @param {object} exec
* @param {{ dshIm?: object }} deps
*/
export async function resolveCreateOrigin(args = {}, exec = {}, deps = {}) {
const explicit = normalizeOrigin(args.origin)
if (explicit) return explicit
const sessionId = typeof args.origin_session_id === 'string'
? args.origin_session_id.trim()
: (typeof args.originSessionId === 'string' ? args.originSessionId.trim() : '')
const callerId = sessionId || callerSessionId(exec)
if (!callerId) return null
const dshIm = deps.dshIm
if (dshIm && typeof dshIm.resolveSessionPeer === 'function') {
try {
const peer = await dshIm.resolveSessionPeer(callerId)
const slim = slimPeer(peer)
if (slim) {
return normalizeOrigin({ kind: 'im', sessionId: callerId, peer: slim })
}
} catch {
// fall through to web origin
}
}
return normalizeOrigin({ kind: 'web', sessionId: callerId })
}
/**
* Resolve delivery for cron_create: explicit args win; else IM peer → matching
* or auto-created 投递目标; else dsh.
@ -301,6 +386,17 @@ export async function resolveCreateDelivery(args = {}, exec = {}, deps = {}) {
const IM_MAX_CHARS = 3500
/**
* Shared body for IM delivery and session mirror (without @mention prefix).
* @param {object} job
* @param {string} [summary]
*/
export function formatRunResultBody(job, summary) {
const text = String(summary || '').trim() || `(定时任务「${job?.name || ''}」已完成,无文本摘要)`
const clipped = text.length > IM_MAX_CHARS ? `${text.slice(0, IM_MAX_CHARS)}…` : text
return `【定时任务 · ${job?.name || ''}】\n${clipped}`
}
/**
* Send run summary to IM when job.delivery.kind === 'im'.
* Group deliveries @mention the creator when mentionJid was captured at create time.
@ -314,17 +410,165 @@ export async function deliverRunToIm(job, summary, deps = {}) {
error.code = 'IM_UNAVAILABLE'
throw error
}
const text = String(summary || '').trim() || `(定时任务「${job.name}」已完成,无文本摘要)`
const clipped = text.length > IM_MAX_CHARS ? `${text.slice(0, IM_MAX_CHARS)}…` : text
const header = `【定时任务 · ${job.name}】\n`
const headerAndBody = formatRunResultBody(job, summary)
const mentionJid = typeof delivery.mentionJid === 'string' ? delivery.mentionJid.trim() : ''
let body = `${header}${clipped}`
let body = headerAndBody
const options = {}
if (mentionJid && /@(s\.whatsapp\.net|lid)$/i.test(mentionJid)) {
const token = mentionJid.slice(0, mentionJid.indexOf('@'))
body = `@${token}\n${header}${clipped}`
body = `@${token}\n${headerAndBody}`
options.mentions = [mentionJid]
}
await dshIm.send(delivery.botId, delivery.targetId, body, options)
return { sent: true, mentioned: Boolean(options.mentions) }
}
/**
* Resolve which live/effective session should receive a mirrored run summary.
* Prefer the current IM conversation binding over a stale origin.sessionId.
* @param {object} job
* @param {{ dshIm?: object }} deps
* @returns {Promise<{ sessionId: string, via: string } | null>}
*/
export async function resolveMirrorSession(job, deps = {}) {
const dshIm = deps.dshIm
const origin = normalizeOrigin(job?.origin)
const delivery = job?.delivery
if (origin?.peer?.botId && origin.peer.conversationKey
&& dshIm && typeof dshIm.resolveConversationSession === 'function') {
try {
const hit = await dshIm.resolveConversationSession(
origin.peer.botId,
origin.peer.conversationKey,
)
const sessionId = typeof hit?.sessionId === 'string' ? hit.sessionId.trim() : ''
if (sessionId) return { sessionId, via: 'conversation' }
} catch {
// try target / origin fallbacks
}
}
if (delivery?.kind === 'im' && delivery.botId && delivery.targetId
&& dshIm && typeof dshIm.resolveTargetSession === 'function') {
try {
const hit = await dshIm.resolveTargetSession(delivery.botId, delivery.targetId)
const sessionId = typeof hit?.sessionId === 'string' ? hit.sessionId.trim() : ''
if (sessionId) return { sessionId, via: 'target' }
} catch {
// fall through
}
}
if (origin?.sessionId) return { sessionId: origin.sessionId, via: 'origin' }
return null
}
/**
* Append the run summary into the effective origin/IM session without waking a turn.
* Prefer a live agent; otherwise resume the persisted session (never create — that
* id already belongs to the origin chat).
* @param {object} job
* @param {string} [summary]
* @param {{
* dshIm?: object,
* getAgents?: () => object | undefined,
* pluginName?: string,
* newId?: () => string,
* }} deps
*/
export async function mirrorRunToSession(job, summary, deps = {}) {
const target = await resolveMirrorSession(job, deps)
if (!target?.sessionId) {
return { mirrored: false, skipped: true, reason: 'no_target' }
}
const agents = typeof deps.getAgents === 'function' ? deps.getAgents() : undefined
if (!agents) {
return {
mirrored: false,
skipped: true,
reason: 'agents_unavailable',
sessionId: target.sessionId,
via: target.via,
}
}
let agent = typeof agents.get === 'function' ? agents.get(target.sessionId) : null
let resumed = false
if (!agent && typeof agents.resume === 'function') {
try {
const handle = await agents.resume({ resumeSessionId: target.sessionId })
agent = handle?.agent || null
resumed = Boolean(agent)
} catch (error) {
return {
mirrored: false,
skipped: true,
reason: 'session_unavailable',
sessionId: target.sessionId,
via: target.via,
error: error instanceof Error ? error.message : String(error),
}
}
}
if (!agent) {
return {
mirrored: false,
skipped: true,
reason: 'session_unavailable',
sessionId: target.sessionId,
via: target.via,
}
}
const text = formatRunResultBody(job, summary)
const plugin = typeof deps.pluginName === 'string' && deps.pluginName.trim()
? deps.pluginName.trim()
: 'dsh-ops-cron'
const id = typeof deps.newId === 'function' ? deps.newId() : `cron-mirror-${Date.now()}`
const message = {
id,
role: 'user',
content: [{ type: 'text', text }],
source: { kind: 'plugin', plugin },
}
try {
if (typeof agent.session?.append === 'function') {
agent.session.append('user/message', message, { surfaceOp: 'append' })
return {
mirrored: true,
sessionId: target.sessionId,
via: target.via,
method: 'append',
resumed,
}
}
if (typeof agent.inject === 'function') {
agent.inject(message)
return {
mirrored: true,
sessionId: target.sessionId,
via: target.via,
method: 'inject',
resumed,
}
}
} catch (error) {
return {
mirrored: false,
skipped: true,
reason: 'mirror_failed',
sessionId: target.sessionId,
via: target.via,
error: error instanceof Error ? error.message : String(error),
}
}
return {
mirrored: false,
skipped: true,
reason: 'no_append',
sessionId: target.sessionId,
via: target.via,
}
}

View file

@ -65,6 +65,7 @@ export function publicJob(job) {
...job.delivery.mentionName ? { mentionName: job.delivery.mentionName } : {},
}
: { kind: 'dsh' },
...job.origin ? { origin: job.origin } : {},
schedule: job.schedule,
createdAt: job.createdAt,
updatedAt: job.updatedAt,

View file

@ -8,7 +8,7 @@ import { mkdir } from 'node:fs/promises'
import { homedir } from 'node:os'
import { basename, join } from 'node:path'
import { claimOccurrence, executeClaimedRun, extractAssistantText, interruptActiveRuns, publicJob, settleRun, TITLE_PREFIX } from './fire.js'
import { deliverRunToIm, mergeDeliveryMention } from './delivery.js'
import { deliverRunToIm, mergeDeliveryMention, mirrorRunToSession, normalizeOrigin } from './delivery.js'
import { workspaceVisibleIds } from './isolation.js'
import { decideDispatch, nextFire, validateSchedule } from './scheduler.js'
import {
@ -112,6 +112,7 @@ function runView(run) {
* @param {string} [options.filePath]
* @param {object} [options.sessionPort] { createAndPrompt, archiveSession, waitForTurn }
* @param {() => object|undefined} [options.getDshIm] soft-injected proactive IM API
* @param {() => object|undefined} [options.getAgents] soft-injected agents service for session mirror
* @param {{ warn?: Function, info?: Function }} [options.logger]
* @param {number} [options.tickIntervalMs]
*/
@ -121,6 +122,7 @@ export function createHostService(options = {}) {
const store = options.store || createStore({ filePath })
const sessionPort = options.sessionPort || null
const getDshIm = typeof options.getDshIm === 'function' ? options.getDshIm : () => undefined
const getAgents = typeof options.getAgents === 'function' ? options.getAgents : () => undefined
const logger = options.logger || console
const tickIntervalMs = Number(options.tickIntervalMs) > 0 ? Number(options.tickIntervalMs) : 15_000
let timer = null
@ -144,6 +146,36 @@ export function createHostService(options = {}) {
}
}
async function maybeMirrorResult(job, run) {
if (!job || !run || run.status === 'skipped') return
try {
const result = await mirrorRunToSession(job, run.summary || run.error || '', {
dshIm: getDshIm(),
getAgents,
pluginName: PLUGIN_NAME,
newId: () => randomUUID(),
})
if (result?.mirrored) {
logger.info?.(`[dsh-ops-cron] mirrored run ${run.id} → session ${result.sessionId} via ${result.via} (${result.method}${result.resumed ? ', resumed' : ''})`)
} else if (result?.reason && result.reason !== 'no_target') {
logger.warn?.(`[dsh-ops-cron] mirror skipped for run ${run.id}: ${result.reason}${result.sessionId ? ` session=${result.sessionId}` : ''}${result.error ? ` (${result.error})` : ''}`)
} else if (!result?.skipped) {
logger.warn?.(`[dsh-ops-cron] mirror incomplete for run ${run.id}: ${result?.reason || 'unknown'}`)
}
} catch (error) {
logger.warn?.(`[dsh-ops-cron] mirror failed for job ${job.id}: ${error instanceof Error ? error.message : error}`)
}
}
async function settleAndNotify(jobId, runId, claimedDecision) {
const state = await snapshot()
const run = (state.runs || []).find((row) => row.id === runId)
const job = getJob(state, jobId)
await maybeDeliverIm(job, run)
await maybeMirrorResult(job, run)
return { job: jobView(job), run: runView(run), decision: claimedDecision }
}
async function createJob(input) {
const t = now()
let created
@ -178,11 +210,15 @@ export function createHostService(options = {}) {
delivery: patch.delivery !== undefined
? mergeDeliveryMention(job.delivery, patch.delivery)
: job.delivery,
origin: patch.origin !== undefined
? (normalizeOrigin(patch.origin) || undefined)
: job.origin,
}
const record = createJobRecord(nextInput, current, t)
record.createdAt = job.createdAt
record.lastRunAt = job.lastRunAt
record.lastStatus = job.lastStatus
if (patch.origin === undefined && job.origin) record.origin = job.origin
if (patch.schedule === undefined && patch.enabled === undefined) {
record.nextRunAt = job.nextRunAt
}
@ -246,13 +282,9 @@ export function createHostService(options = {}) {
}
const settled = await store.mutate((current) => settleRun(current, run.id, terminal, now()))
run = (settled.runs || []).find((row) => row.id === claimed.run.id)
const job = getJob(settled, jobId)
await maybeDeliverIm(job, run)
return { job: jobView(job), run: runView(run), decision: claimed.decision }
return settleAndNotify(jobId, run.id, claimed.decision)
}
const job = getJob(executed, jobId)
await maybeDeliverIm(job, run)
return { job: jobView(job), run: runView(run), decision: claimed.decision }
return settleAndNotify(jobId, claimed.run.id, claimed.decision)
}
async function tick() {

5
lib/index.d.ts vendored
View file

@ -42,5 +42,10 @@ export const TITLE_PREFIX: string
export function claimOccurrence(state: object, jobId: string, now: number, trigger: string, policies?: object): object
export function executeClaimedRun(state: object, runId: string, deps: object): Promise<object>
export function normalizeDelivery(input?: object): { kind: 'dsh' | 'im', botId?: string, targetId?: string }
export function normalizeOrigin(input?: object): { kind: 'web' | 'im', sessionId?: string, peer?: object } | null
export function resolveCreateDelivery(args?: object, exec?: object, deps?: object): Promise<{ kind: 'dsh' | 'im', botId?: string, targetId?: string }>
export function resolveCreateOrigin(args?: object, exec?: object, deps?: object): Promise<{ kind: 'web' | 'im', sessionId?: string, peer?: object } | null>
export function resolveMirrorSession(job: object, deps?: object): Promise<{ sessionId: string, via: string } | null>
export function formatRunResultBody(job: object, summary?: string): string
export function deliverRunToIm(job: object, summary?: string, deps?: object): Promise<object>
export function mirrorRunToSession(job: object, summary?: string, deps?: object): Promise<object>

View file

@ -51,8 +51,13 @@ export { normalizeJobModel } from './store.js'
export { claimOccurrence, executeClaimedRun, extractAssistantText, TITLE_PREFIX } from './fire.js'
export {
deliverRunToIm,
formatRunResultBody,
mirrorRunToSession,
normalizeDelivery,
normalizeOrigin,
resolveCreateDelivery,
resolveCreateOrigin,
resolveMirrorSession,
} from './delivery.js'
function loadPkg(id) {
@ -131,10 +136,12 @@ function registerWebRoute(ctx, handler) {
export function apply(ctx, config = {}) {
const entry = resolveConfig(config)
const getDshIm = () => tryGet(ctx, 'dshIm')
const getAgents = () => tryGet(ctx, 'agents')
const getAgentPresets = () => tryGet(ctx, 'agentPresets')
const service = createHostService({
sessionPort: makeLiveSessionPort(ctx),
getDshIm,
getAgents,
logger: ctx.logger,
})

View file

@ -8,7 +8,7 @@ import { dirname, join } from 'node:path'
import { randomUUID } from 'node:crypto'
import { applyRunIsolation, recordHiddenSession } from './isolation.js'
import { nextFire, validateSchedule } from './scheduler.js'
import { normalizeDelivery } from './delivery.js'
import { normalizeDelivery, normalizeOrigin } from './delivery.js'
import { normalizeAgentPresetId } from './preset.js'
export const STORE_VERSION = 1
@ -95,6 +95,7 @@ export function createJobRecord(input, state, now) {
const timeoutMinutes = Number(input.timeoutMinutes)
const model = normalizeJobModel(input)
const delivery = normalizeDelivery(input.delivery)
const origin = normalizeOrigin(input.origin)
const agentPreset = normalizeAgentPresetId(input.agentPreset ?? input.agent_preset)
const job = {
id: String(input.id || newId()),
@ -110,6 +111,7 @@ export function createJobRecord(input, state, now) {
reasoningEffort: model.reasoningEffort,
agentPreset,
delivery,
...origin ? { origin } : {},
schedule: schedule.kind === 'cron'
? { kind: 'cron', expr: schedule.expr, timezone: schedule.timezone }
: { kind: 'at', at: new Date(schedule.at).toISOString(), timezone: schedule.timezone },

View file

@ -4,7 +4,7 @@
*/
import { formatInZone, resolveTodayAt } from './scheduler.js'
import { deliveryLine, resolveCreateDelivery } from './delivery.js'
import { deliveryLine, resolveCreateDelivery, resolveCreateOrigin } from './delivery.js'
import { resolveCreateAgentPreset } from './preset.js'
const JOB_SCHEMA = {
@ -22,6 +22,7 @@ const JOB_SCHEMA = {
reasoningEffort: { type: 'string' },
agentPreset: { type: 'string' },
delivery: { type: 'object', additionalProperties: true },
origin: { type: 'object', additionalProperties: true },
schedule: { type: 'object', additionalProperties: true },
createdAt: { type: 'number' },
updatedAt: { type: 'number' },
@ -204,6 +205,7 @@ export function cronToolDefinitions(service, deps = {}) {
const timeout = Number(args.timeout_minutes)
try {
const delivery = await resolveCreateDelivery(args, exec, { dshIm: getDshIm() })
const origin = await resolveCreateOrigin(args, exec, { dshIm: getDshIm() })
const agentPreset = await resolveCreateAgentPreset(args, exec, {
dshIm: getDshIm(),
agentPresets: getAgentPresets(),
@ -217,6 +219,7 @@ export function cronToolDefinitions(service, deps = {}) {
timeoutMinutes: Number.isFinite(timeout) && timeout > 0 ? timeout : undefined,
enabled: args.enabled !== false,
delivery,
...origin ? { origin } : {},
agentPreset,
})
return { job }