import Router from '@koa/router' import {randomUUID} from 'node:crypto' import {z} from 'zod' import {parse,type WorkspaceStore} from './workspaceStore.js' import type {LocalAuthStore} from './localAuth.js' import type {WorkspaceNodes} from './workspaceGateway.js' import type {WorkspaceRuntime} from './workspaceRuntime.js' import type {WorkspaceAgent,WorkspaceConversation,WorkspaceRun} from '../shared/workspace.js' import type {WorkspaceRoutine,WorkspaceSchedule,WorkspaceRoutineRun} from '../shared/workspacePanels.js' import {HttpError} from './errors.js' const timezone=z.string().max(100).refine(value=>{try{new Intl.DateTimeFormat('en',{timeZone:value});return true}catch{return false}},'时区无效') const clock=z.string().regex(/^([01]\d|2[0-3]):[0-5]\d$/) const schedule=z.discriminatedUnion('kind',[ z.object({kind:z.literal('once'),timezone,at:z.number().int().positive()}).strict(), z.object({kind:z.literal('interval'),timezone,everyMinutes:z.number().int().min(1).max(525600)}).strict(), z.object({kind:z.literal('daily'),timezone,time:clock}).strict(), z.object({kind:z.literal('weekly'),timezone,time:clock,weekdays:z.array(z.number().int().min(0).max(6)).min(1).max(7)}).strict(), ]) const input=z.object({name:z.string().trim().min(1).max(100),prompt:z.string().trim().min(1).max(24000),enabled:z.boolean(),schedule}).strict() export function nextRoutineAt(s:WorkspaceSchedule,after:number):number|undefined { if(s.kind==='once')return s.at!>after?s.at:undefined if(s.kind==='interval')return after+s.everyMinutes!*60000 const formatter=new Intl.DateTimeFormat('en-US',{timeZone:s.timezone,hourCycle:'h23',hour:'2-digit',minute:'2-digit',weekday:'short'}) for(let at=Math.floor(after/60000)*60000+60000;at<=after+8*86400000;at+=60000){ const parts=Object.fromEntries(formatter.formatToParts(at).map(p=>[p.type,p.value])) if(`${parts.hour}:${parts.minute}`===s.time&&(s.kind==='daily'||s.weekdays!.includes(['Sun','Mon','Tue','Wed','Thu','Fri','Sat'].indexOf(parts.weekday!))))return at } throw new HttpError(400,'无法计算下次执行时间','routine_schedule_invalid') } const terminal=(status:string)=>['complete','failed','interrupted','skipped'].includes(status) export class WorkspaceRoutines { private timer?:ReturnType;private ticking=false;private closed=false constructor(readonly store:WorkspaceStore,readonly auth:LocalAuthStore,readonly nodes:WorkspaceNodes,readonly runtime:WorkspaceRuntime){} private agent(owner:string,id:string){const agent=this.store.require(owner,'agent',id);if(agent.archived||agent.temporaryGoalId||agent.remoteAgentId)throw new HttpError(409,'此机器人不能建立定时任务','routine_agent_unavailable');this.nodes.requireSource(owner,agent);return agent} save(owner:string,agentId:string,body:unknown,id:string=randomUUID()){ this.agent(owner,agentId);const value=parse(input,body),old=this.store.get(owner,'routine',id) if(old&&old.agentId!==agentId)throw new HttpError(404,'定时任务不存在','not_found') const now=Date.now(),unchanged=!!old&&old.enabled===value.enabled&&JSON.stringify(parse(schedule,old.schedule))===JSON.stringify(value.schedule),nextAt=unchanged?old!.nextAt:value.enabled?nextRoutineAt(value.schedule,now):undefined if(value.enabled&&!nextAt)throw new HttpError(400,'请选择未来的执行时间','routine_time_invalid') const routine:WorkspaceRoutine={...value,id,agentId,nextAt,createdAt:old?.createdAt??now,updatedAt:now,lastAt:old?.lastAt} this.store.put(owner,'routine',id,routine);this.store.event(owner,'routine.changed',routine);return routine } runs(owner:string,agentId:string){return this.store.list(owner,'routine-run').filter(r=>r.agentId===agentId).map(r=>{ const run=r.runId?this.store.get(owner,'run',r.runId):undefined return run?{...r,status:run.status,error:run.error}:r }).sort((a,b)=>b.startedAt-a.startedAt).slice(0,100)} run(owner:string,routine:WorkspaceRoutine,scheduledAt:number,requestId:string=randomUUID()){ return this.store.command(owner,requestId,{operation:'routine.run',routineId:routine.id,scheduledAt},()=>this.store.atomic(()=>{ this.agent(owner,routine.agentId) const id=randomUUID(),entry:WorkspaceRoutineRun={id,routineId:routine.id,agentId:routine.agentId,scheduledAt,startedAt:Date.now(),status:'queued'} try { if(this.store.list(owner,'routine-run').some(r=>r.routineId===routine.id&&!terminal((r.runId?this.store.get(owner,'run',r.runId)?.status:undefined)??r.status)))throw new HttpError(409,'上次定时任务仍在执行,本次已跳过','routine_overlap') const conversation=this.store.list(owner,'conversation').find(c=>c.kind==='direct'&&!c.archived&&c.memberIds[0]===routine.agentId) if(!conversation)throw new HttpError(409,'机器人的聊天已归档或删除','routine_conversation_unavailable') const run=this.runtime.send(owner,conversation.id,{requestId:id,content:routine.prompt}) Object.assign(entry,{runId:run.id,conversationId:conversation.id}) }catch(error){entry.status=error instanceof HttpError&&error.code==='routine_overlap'?'skipped':'failed';entry.error=error instanceof Error?error.message:'定时任务未能启动'} this.store.put(owner,'routine-run',id,entry) const old=this.store.list(owner,'routine-run').filter(r=>r.agentId===routine.agentId).sort((a,b)=>b.startedAt-a.startedAt).slice(100) for(const run of old)if(terminal(this.store.get(owner,'run',run.runId??'')?.status??run.status))this.store.remove(owner,'routine-run',run.id) return entry })) } tick(now=Date.now()){ if(this.closed||this.ticking)return;this.ticking=true try {for(const owner of this.store.owners())for(const routine of this.store.list(owner,'routine')){ if(!routine.enabled||!routine.nextAt||routine.nextAt>now)continue this.store.atomic(()=>{ const due=routine.nextAt! try{if(!this.auth.isUserActive(owner))throw new Error('账号当前不可用');this.run(owner,routine,due)}catch(error){const id=randomUUID();routine.enabled=false;this.store.put(owner,'routine-run',id,{id,routineId:routine.id,agentId:routine.agentId,scheduledAt:due,startedAt:now,status:'failed',error:error instanceof Error?error.message:'定时任务未能启动'})} routine.lastAt=due;routine.nextAt=routine.enabled?nextRoutineAt(routine.schedule,now):undefined;if(!routine.nextAt)routine.enabled=false routine.updatedAt=now;this.store.put(owner,'routine',routine.id,routine);this.store.event(owner,'routine.changed',routine) }) }}finally{this.ticking=false} } start(){if(this.timer)return;this.tick();this.timer=setInterval(()=>this.tick(),10000);this.timer.unref()} close(){this.closed=true;clearInterval(this.timer)} router(){const router=new Router() router.get('/api/app/agents/:id/routines',ctx=>{const owner=this.auth.require(ctx).id;this.store.require(owner,'agent',ctx.params.id);ctx.body={routines:this.store.list(owner,'routine').filter(r=>r.agentId===ctx.params.id),runs:this.runs(owner,ctx.params.id)}}) router.post('/api/app/agents/:id/routines',ctx=>{ctx.body={routine:this.save(this.auth.require(ctx).id,ctx.params.id,(ctx.request as any).body)};ctx.status=201}) router.put('/api/app/agents/:id/routines/:routineId',ctx=>{const owner=this.auth.require(ctx).id;this.store.require(owner,'routine',ctx.params.routineId);ctx.body={routine:this.save(owner,ctx.params.id,(ctx.request as any).body,ctx.params.routineId)}}) router.delete('/api/app/agents/:id/routines/:routineId',ctx=>{const owner=this.auth.require(ctx).id,r=this.store.require(owner,'routine',ctx.params.routineId);if(r.agentId!==ctx.params.id)throw new HttpError(404,'定时任务不存在','not_found');this.store.remove(owner,'routine',r.id);ctx.body={ok:true}}) router.post('/api/app/agents/:id/routines/:routineId/run',ctx=>{const owner=this.auth.require(ctx).id,r=this.store.require(owner,'routine',ctx.params.routineId);if(r.agentId!==ctx.params.id)throw new HttpError(404,'定时任务不存在','not_found');const body=parse(z.object({requestId:z.string().uuid()}).strict(),(ctx.request as any).body);ctx.body={run:this.run(owner,r,0,body.requestId)}}) return router } }