Permissions

apps/server/src/index.ts

@@ -28,18 +28,16 @@ import { getPinoTransport } from '@hyperdx/node-opentelemetry'
 import { PRODUCTION } from '@magickml/config'
 
 if (PRODUCTION) {
-	initLogger({
-		name: 'cloud-agent-worker',
-		transport: {
-			targets: [
-				getPinoTransport('info')
-			]
-		},
-		level: 'info',
-	})
+  initLogger({
+    name: 'cloud-agent-worker',
+    transport: {
+      targets: [getPinoTransport('info')]
+    },
+    level: 'info'
+  })
 } else {
-	initLogger({ name: 'cloud-agent-worker' })
-}	
+  initLogger({ name: 'cloud-agent-worker' })
+}
 const logger = getLogger()
 
 // log handle errors
@@ -69,7 +67,7 @@ const routes: Route[] = [...spells, ...apis, ...serverRoutes]
  * form and multipart-json requests, and routes.
  */
 async function init() {
-  await initApp()
+  await initApp('server')
   await initAgentCommander()
   // load plugins
   await (async () => {

package.json

@@ -125,6 +125,7 @@
     "ethers": "5.7.2",
     "expletives": "0.1.5",
     "express": "4.18.2",
+    "feathers-permissions": "^2.1.4",
     "feathers-sync": "3.0.3",
     "flatted": "3.2.7",
     "flexlayout-react": "0.7.7",

packages/agents/src/lib/Agent.ts

@@ -201,6 +201,7 @@ export class Agent implements AgentInterface {
 
   async runWorker(job: Job<AgentRunJob>) {
     // the job name is the agent id.  Only run if the agent id matches.
+    this.logger.debug('running worker', { id: this.id, data: job.data })
     if (this.id !== job.data.agentId) return
 
     const { data } = job
@@ -235,6 +236,7 @@ export class Agent implements AgentInterface {
           ...this.secrets,
           ...data.secrets,
         },
+        sessionId: data?.sessionId,
         publicVariables: this.publicVariables,
         runSubspell: data.runSubspell,
         app,
@@ -269,6 +271,7 @@ export class Agent implements AgentInterface {
 
 export interface AgentRunJob {
   inputs: MagickSpellInput
+  sessionId?: string
   jobId: string
   agentId: string
   spellId: string

packages/agents/src/lib/AgentCommander.ts

@@ -15,7 +15,8 @@ import { AgentResult, AgentRunJob } from './Agent'
 import { AGENT_RESPONSE_TIMEOUT_MSEC } from '@magickml/config'
 
 export type RunRootSpellArgs = {
-  agent: Agent
+  agent?: Agent
+  agentId?: string
   inputs: MagickSpellInput
   componentName?: string
   runSubspell?: boolean
@@ -25,6 +26,7 @@ export type RunRootSpellArgs = {
   isSubSpell?: boolean
   currentJob?: Job<AgentRunJob>
   subSpellDepth?: number
+  sessionId?: string
 }
 
 interface AgentCommanderArgs {
@@ -41,7 +43,10 @@ export class AgentCommander extends EventEmitter {
   }
 
   runSpellWithResponse(args: RunRootSpellArgs) {
-    const { agent } = args
+    const { agentId, agent } = args
+    const id = agentId || agent?.id
+    if (!id) throw new Error('Agent or agent id is required')
+
     return new Promise((resolve, reject) => {
       ;(async () => {
         setTimeout(() => {
@@ -50,7 +55,7 @@ export class AgentCommander extends EventEmitter {
 
         let jobId: null | string = null
 
-        const agentMessageName = AGENT_RUN_RESULT(agent.id)
+        const agentMessageName = AGENT_RUN_RESULT(id)
 
         this.pubSub.subscribe(agentMessageName, (data: AgentResult) => {
           if (data.result.error) {
@@ -66,7 +71,7 @@ export class AgentCommander extends EventEmitter {
           }
         })
 
-        const agentErrorName = AGENT_RUN_ERROR(agent.id)
+        const agentErrorName = AGENT_RUN_ERROR(id)
         this.pubSub.subscribe(agentErrorName, (data: AgentResult) => {
           if (data.jobId === jobId) {
             this.pubSub.unsubscribe(agentErrorName)
@@ -83,46 +88,54 @@ export class AgentCommander extends EventEmitter {
   private runRootSpellArgsToString(
     jobId: string,
     {
-      agent,
+      agentId,
       inputs,
       componentName,
       runSubspell,
       secrets,
       publicVariables,
       spellId,
       subSpellDepth,
+      sessionId,
     }: RunRootSpellArgs
   ) {
     return JSON.stringify({
       jobId,
-      agentId: agent.id,
-      spellId: spellId || agent.rootSpellId,
+      agentId,
+      spellId: spellId,
       inputs,
       componentName,
       runSubspell,
       secrets,
       publicVariables,
       subSpellDepth,
+      sessionId,
     })
   }
 
   async runSubSpell(args: RunRootSpellArgs) {
-    const { agent } = args
+    const { agentId, agent } = args
+    const id = agentId || agent?.id
+    if (!id) throw new Error('Agent or agent id is required')
+
     const jobId = uuidv4()
     await this.pubSub.publish(
-      AGENT_RUN_JOB(agent.id),
+      AGENT_RUN_JOB(id),
       this.runRootSpellArgsToString(jobId, args)
     )
     return jobId
   }
 
   async runSpell(args: RunRootSpellArgs) {
-    const { agent } = args
-    this.logger.debug(`Running Spell on Agent: ${agent.id}`)
-    this.logger.debug(AGENT_RUN_JOB(agent.id))
+    const { agent, agentId } = args
+    const id = agentId || agent?.id
+    if (!id) throw new Error('Agent or agent id is required')
+
+    this.logger.debug(`Running Spell on Agent: ${id}`)
+    this.logger.debug(AGENT_RUN_JOB(id))
     const jobId = uuidv4()
     await this.pubSub.publish(
-      AGENT_RUN_JOB(agent.id),
+      AGENT_RUN_JOB(id),
       this.runRootSpellArgsToString(jobId, args)
     )
     return jobId

packages/cloud-agent-manager/src/lib/CloudAgentManager.ts

@@ -1,10 +1,15 @@
-import pino from "pino"
-import { diff, unique } from "radash"
-import { AGENT_DELETE, AGENT_DELETE_JOB, AGENT_UPDATE_JOB, getLogger } from "@magickml/core"
-import type { Reporter } from "./Reporters"
-import { type PubSub, type MessageQueue, app } from "@magickml/server-core"
-import { Agent } from "packages/core/server/src/services/agents/agents.schema"
-import { HEARTBEAT_MSEC, MANAGER_WARM_UP_MSEC } from "@magickml/config"
+import pino from 'pino'
+import { diff, unique } from 'radash'
+import {
+  AGENT_DELETE,
+  AGENT_DELETE_JOB,
+  AGENT_UPDATE_JOB,
+  getLogger,
+} from '@magickml/core'
+import type { Reporter } from './Reporters'
+import { type PubSub, type MessageQueue, app } from '@magickml/server-core'
+import { Agent } from 'packages/core/server/src/services/agents/agents.schema'
+import { HEARTBEAT_MSEC, MANAGER_WARM_UP_MSEC } from '@magickml/config'
 
 interface CloudAgentManagerConstructor {
   pubSub: PubSub
@@ -42,69 +47,92 @@ export class CloudAgentManager {
       this.logger.info(`Agent Updated: ${agent.id}`)
 
       if (agent.enabled) {
-        this.logger.info(`Agent ${agent.id} enabled, adding to cloud agent worker`)
-        await this.newQueue.addJob('agent:new', {agentId: agent.id})
+        this.logger.info(
+          `Agent ${agent.id} enabled, adding to cloud agent worker`
+        )
+        await this.newQueue.addJob('agent:new', { agentId: agent.id })
         this.logger.debug(`Agent create job for ${agent.id} added`)
         return
       }
 
-      this.pubSub.publish(AGENT_UPDATE_JOB(agent.id), JSON.stringify({ agentId: agent.id }))
+      this.pubSub.publish(
+        AGENT_UPDATE_JOB(agent.id),
+        JSON.stringify({ agentId: agent.id })
+      )
     })
 
     this.agentStateReporter.on(AGENT_DELETE, async (data: unknown) => {
       const agent = data as Agent
-      this.pubSub.publish(AGENT_DELETE_JOB(agent.id), JSON.stringify({ agentId: agent.id }))
+      this.pubSub.publish(
+        AGENT_DELETE_JOB(agent.id),
+        JSON.stringify({ agentId: agent.id })
+      )
     })
   }
 
   async dedupeAgents(agents: string[]) {
     const deduped = unique(agents)
     const diffAgents = diff(agents, deduped)
 
-    this.logger.info("deduping agents %o", diffAgents)
-    diffAgents.forEach(async (agentId) => {
-      await this.pubSub.publish(AGENT_DELETE_JOB(agentId), JSON.stringify({ agentId: agentId }))
-      await this.newQueue.addJob('agent:new', {agentId: agentId})
+    this.logger.trace('deduping agents %o', diffAgents)
+    diffAgents.forEach(async agentId => {
+      await this.pubSub.publish(
+        AGENT_DELETE_JOB(agentId),
+        JSON.stringify({ agentId: agentId })
+      )
+      await this.newQueue.addJob('agent:new', { agentId: agentId })
     })
 
     return deduped
   }
 
   // Eventually we'll need this heartbeat to keep track of running agents on workers
   async heartbeat() {
-    this.logger.debug("Started heartbeat")
+    this.logger.debug('Started heartbeat')
     let agentsOfWorkers: string[] = []
-    this.pubSub.subscribe("heartbeat-pong", async (agents: string[]) => {
-      this.logger.debug("Got heartbeat pong")
+    this.pubSub.subscribe('heartbeat-pong', async (agents: string[]) => {
+      this.logger.trace('Got heartbeat pong')
       agents.forEach(a => agentsOfWorkers.push(a))
       agentsOfWorkers = await this.dedupeAgents(agentsOfWorkers)
     })
-    await this.pubSub.publish("heartbeat-ping", "{}")
-
-    setTimeout(() => 
-      setInterval(async () => {
-        this.logger.debug(`Starting Heartbeat update`)
-        const enabledAgents = await app.service('agents').find({
-          query: {
-            enabled: true,
-          },
-        })
-
-        const agentDiff = diff(enabledAgents.data.map(a => a.id), Array.from(agentsOfWorkers))
-        const agentsToUpdate = enabledAgents.data.filter(a => agentDiff.includes(a.id))
-
-        if (agentDiff.length > 0) {
-          this.logger.info(`Found ${agentDiff.length} agents to Update`)
-          const agentPromises: Promise<any>[] = []
-          for (const agent of agentsToUpdate) {
-            this.logger.debug(`Adding agent ${agent.id} to cloud agent worker`)
-            agentPromises.push(this.newQueue.addJob('agent:new', {agentId: agent.id}))
+    await this.pubSub.publish('heartbeat-ping', '{}')
+
+    setTimeout(
+      () =>
+        setInterval(async () => {
+          this.logger.trace(`Starting Heartbeat update`)
+          const enabledAgents = await app.service('agents').find({
+            query: {
+              enabled: true,
+            },
+          })
+
+          const agentDiff = diff(
+            enabledAgents.data.map(a => a.id),
+            Array.from(agentsOfWorkers)
+          )
+          const agentsToUpdate = enabledAgents.data.filter(a =>
+            agentDiff.includes(a.id)
+          )
+
+          if (agentDiff.length > 0) {
+            this.logger.info(`Found ${agentDiff.length} agents to Update`)
+            const agentPromises: Promise<any>[] = []
+            for (const agent of agentsToUpdate) {
+              this.logger.debug(
+                `Adding agent ${agent.id} to cloud agent worker`
+              )
+              agentPromises.push(
+                this.newQueue.addJob('agent:new', { agentId: agent.id })
+              )
+            }
+
+            await Promise.all(agentPromises)
           }
-
-          await Promise.all(agentPromises)
-        }
-        agentsOfWorkers = [];
-        this.pubSub.publish("heartbeat-ping", "{}")
-      }, HEARTBEAT_MSEC), MANAGER_WARM_UP_MSEC)
+          agentsOfWorkers = []
+          this.pubSub.publish('heartbeat-ping', '{}')
+        }, HEARTBEAT_MSEC),
+      MANAGER_WARM_UP_MSEC
+    )
   }
 }

packages/cloud-agent-worker/src/lib/cloud-agent-worker.ts

@@ -40,7 +40,7 @@ export class CloudAgentWorker extends AgentManager {
     })
 
     this.pubSub.subscribe('heartbeat-ping', async () => {
-      this.logger.debug('Got heartbeat ping')
+      this.logger.trace('Got heartbeat ping')
       const agentIds = Object.keys(this.currentAgents)
       this.pubSub.publish('heartbeat-pong', JSON.stringify(agentIds))
     })