Task Queue & Orchestration

Redis — shared queues (all tenants)
├── agent__client-researcher          concurrency: variable
├── agent__competitor-researcher      concurrency: variable
├── agent__context-file-writer        concurrency: variable
├── agent__strategy-writer            concurrency: variable
├── agent__deliverable-planner        concurrency: variable
├── agent__activity-planner           concurrency: variable
├── agent__keyword-researcher         concurrency: variable
├── agent__content-brief-writer       concurrency: variable
├── agent__site-auditor               concurrency: variable
├── agent__backlink-researcher        concurrency: variable
├── agent__backlink-outreach-writer   concurrency: variable
├── agent__blog-writer                concurrency: variable
├── agent__gbp-post-writer            concurrency: variable
├── agent__social-post-writer         concurrency: variable
├── agent__email-writer               concurrency: variable
├── agent__landing-page-writer        concurrency: variable
├── agent__google-ads-writer          concurrency: variable
├── agent__meta-ads-writer            concurrency: variable
├── agent__social-calendar-planner    concurrency: variable
├── agent__ads-analyst                concurrency: variable
├── agent__report-writer              concurrency: variable
├── agent__anomaly-detector           concurrency: variable
├── agent__topic-researcher           concurrency: variable
├── agent__research-note-writer       concurrency: variable
├── agent__review-response-writer     concurrency: variable
└── dead-letter                       concurrency: 1  (global alerting)

interface ActivityJobData {
  activityId:    string;        // PostgreSQL activities.id
  tenantId:      string;        // For tenant isolation verification
  agentRole:     AgentRole;     // Which agent queue this came from
  deliverableId: string;        // Parent deliverable
  pipelineStep:  number;        // Which step in the activity template
  inputRef?:     string;        // MongoDB ObjectId of previous step's output
  priority:      number;        // 0 = normal, 1 = high, 2 = urgent
 
  // Wake reason — tells the agent why it was dispatched
  wakeReason:    'new_task' | 'rejection' | 'scheduled' | 'unblocked'
               | 'subtask_completed' | 'review_approved' | 'review_feedback';
 
  // Populated only when wakeReason === 'rejection'
  rejectionRef?:  string;       // MongoDB ObjectId of the output that was rejected
  reviewerNotes?: string;       // Human reviewer's written feedback
 
  // Populated only when wakeReason === 'unblocked'
  blockerResolution?: string;   // Human's note explaining how the blocker was resolved
 
  // Populated only when wakeReason === 'subtask_completed'
  subtaskOutputRef?:  string;   // MongoDB ObjectId of the completed subtask's output
  subtaskName?:       string;   // Name of the subtask that completed
 
  // Populated only when wakeReason === 'review_approved' | 'review_feedback'
  reviewApproved?:    boolean;  // true = approved, false = feedback given
  reviewerFeedback?:  string;   // Reviewer's comments (present for both approved + feedback)
}

                         ┌─────────┐
         created ──────► │ created │
                         └────┬────┘
                              │ worker picks up job
                         ┌────▼─────┐
                         │ assigned │
                         └────┬─────┘
                              │ adapter dispatches
                         ┌────▼──────────┐◄──── re-enqueued (any wakeReason)
                         │  in_progress  │
                         └──┬───┬───┬───┘
              ┌─────────────┘   │   └──────────────────────────┐
              │            ┌────┘                              │
     ┌────────▼──────┐  ┌──▼────┐  ┌─────────┐  ┌──────────┐  └──► failed
     │awaiting_      │  │ done  │  │ blocked │  │awaiting_ │       /timeout
     │approval       │  │       │  │         │  │subtask   │          │
     └───────┬───────┘  └───────┘  └────┬────┘  └────┬─────┘     retries?
             │                          │             │
      approved/rejected          blocker        subtask           reassigned
             │                   resolved       completes         (terminal)
      next step /             re-enqueued    re-enqueued
      revision activity       (unblocked)  (subtask_completed)
                                                      │
                                   ┌──────────────────┤
                                   │                  │
                              ┌────▼──────────┐  awaiting_
                              │  in_progress  │  review
                              └───────────────┘      │
                                               approved /
                                               feedback
                                            re-enqueued
                                  (review_approved / review_feedback)

import { Worker } from 'bullmq';
 
function createAgentWorker(agentRole: AgentRole, concurrency: number) {
  const queueName = `agent__${agentRole}`;  // shared across all tenants
 
  return new Worker(
    queueName,
    async (job: Job<ActivityJobData>) => {
      // tenantId is in the payload — worker verifies it before processing
      await processActivity(job.data);
    },
    {
      connection: redisConnection,
      concurrency,
      limiter: {
        max:      10,       // max 10 jobs per minute per worker (runaway protection)
        duration: 60_000,
      },
    }
  );
}

await queue.add(
  'activity',
  jobData,
  {
    // Deduplication key: same activity cannot be queued twice with the same wake reason
    deduplication: {
      id: `${jobData.tenantId}:${jobData.activityId}:${jobData.wakeReason}`,
    },
    priority: jobData.priority,
  }
);

const worker = new Worker(queueName, processor, {
  connection:      getRedisConnection(),
  concurrency:     1,
  lockDuration:    660_000,
  maxStalledCount: 2,  // allow 2 stall-retries before giving up
});

async function processActivity(data: ActivityJobData): Promise<void> {
  // 1. Mark activity as in_progress
  await db.update(activities)
    .set({ status: 'in_progress', startedAt: new Date() })
    .where(and(eq(activities.id, data.activityId), eq(activities.tenantId, data.tenantId)));
 
  // 2. Create activity_run record
  const [run] = await db.insert(activityRuns).values({
    activityId: data.activityId,
    tenantId:   data.tenantId,
    wakeReason: data.wakeReason,
    startedAt:  new Date(),
  }).returning();
 
  // 3. Resolve adapter for this tenant + agent role
  const agentConfig = await getAgentConfig(data.tenantId, data.agentRole);
  const adapter = resolveAdapter(agentConfig);
 
  // 4. Resolve skills — returns client context file (full) + manifest (metadata-only)
  const { contextFile, manifest } = await resolveSkillsForActivity(data.tenantId, data.agentRole);
  const skillsDir = await injectSkills(contextFile, manifest);
 
  // 5. Build task prompt — threads wake reason and rejection context
  const prompt = await buildActivityPrompt(data);
 
  // 6. Resolve the existing session for this campaign + agent role (if any)
  const session = await resolveSession(data.deliverableId, data.agentRole);
 
  try {
    // 7. Dispatch to adapter
    const result = await adapter.dispatch({
      prompt,
      skillsDir,
      sessionId:     session?.externalSessionId ?? undefined,
      activityRunId: run.id,
      wakeReason:    data.wakeReason,
    });
 
    // 8. Persist session state for next heartbeat
    await persistSession(session, result, data.deliverableId, data.agentRole);
 
    // 9. Update run record
    // Note: if the agent called report_blocker, status is already 'blocked' —
    // the tool handler sets it. The adapter returns normally but the activity
    // remains blocked until onBlockerResolved() re-enqueues it.
    const isBlocked = await isActivityBlocked(data.activityId);
    if (!isBlocked) {
      await db.update(activityRuns)
        .set({ status: 'completed', endedAt: new Date(), sessionId: result.sessionId })
        .where(eq(activityRuns.id, run.id));
    }
 
  } catch (err) {
    await db.update(activityRuns)
      .set({ status: 'failed', error: String(err), endedAt: new Date() })
      .where(eq(activityRuns.id, run.id));
    throw err;  // BullMQ catches this and triggers retry
  } finally {
    await cleanupSkillsDir(skillsDir);
  }
}

async function buildActivityPrompt(data: ActivityJobData): Promise<string> {
  const activity = await getActivity(data.activityId, data.tenantId);
  const input = data.inputRef ? await getActivityOutput(data.inputRef) : null;
 
  // Base task description
  let prompt = `# Task: ${activity.name}\n\n${activity.description}`;
 
  // Thread the prior pipeline step's output as context
  if (input) {
    prompt += `\n\n## Input from previous step\n\n${input.content}`;
  }
 
  // Rejection path — agent is revising work a human reviewed and rejected
  if (data.wakeReason === 'rejection') {
    if (data.rejectionRef) {
      const rejected = await getActivityOutput(data.rejectionRef);
      prompt += `\n\n## Your previous output (rejected by reviewer)\n\n${rejected.content}`;
    }
 
    if (data.reviewerNotes) {
      prompt += `\n\n## Reviewer feedback\n\n${data.reviewerNotes}`;
    }
 
    prompt += `\n\nRevise your output to address the reviewer's feedback above. `
            + `Do not restate the feedback — produce the corrected output directly.`;
  }
 
  return prompt;
}

if (data.wakeReason === 'unblocked' && data.blockerResolution) {
  prompt += `\n\n## Blocker resolved\n\n${data.blockerResolution}`;
  prompt += `\n\nResume the task from where you left off.`;
}
 
if (data.wakeReason === 'subtask_completed' && data.subtaskOutputRef) {
  const subtaskOutput = await getActivityOutput(data.subtaskOutputRef);
  prompt += `\n\n## Subtask completed: ${data.subtaskName}\n\n${subtaskOutput.content}`;
  prompt += `\n\nContinue your task using the subtask result above.`;
}
 
if (data.wakeReason === 'review_approved') {
  prompt += `\n\n## Human review: Approved`;
  if (data.reviewerFeedback) prompt += `\n\nReviewer notes: ${data.reviewerFeedback}`;
  prompt += `\n\nYour work was approved. Finalise and complete the task.`;
}
 
if (data.wakeReason === 'review_feedback') {
  prompt += `\n\n## Human review: Feedback provided\n\n${data.reviewerFeedback}`;
  prompt += `\n\nRevise your work based on the feedback above, then complete the task.`;
}

async function onActivityCompleted(activityId: string, tenantId: string): Promise<void> {
  const activity = await getActivity(activityId, tenantId);
  const template = await getActivityTemplate(activity.templateId);
 
  // Find next step in pipeline
  const nextStep = template.steps[activity.pipelineStep + 1];
 
  if (!nextStep) {
    // Pipeline complete — mark deliverable unit done
    await incrementDeliverablePeriodCount(activity.deliverableId, tenantId);
    return;
  }
 
  // Create next activity
  const nextActivity = await db.insert(activities).values({
    tenantId,
    deliverableId:       activity.deliverableId,
    deliverablePeriodId: activity.deliverablePeriodId,
    templateId:          activity.templateId,
    pipelineStep:        activity.pipelineStep + 1,
    name:                nextStep.name,
    assigneeType:        nextStep.assigneeType,
    assigneeAgentRole:   nextStep.agentRole,
    isApprovalGate:      nextStep.isApproval,
    inputRef:            activity.outputRef,  // Output of this step is input to next
    status:              'created',
  }).returning();
 
  // Enqueue if agent activity; surface in inbox if human activity
  if (nextStep.assigneeType === 'agent') {
    await enqueueActivity(nextActivity[0], tenantId, nextStep.agentRole, {
      wakeReason: 'new_task',
    });
  }
  // Human activities are surfaced automatically via the DM Portal inbox query
}

async function onActivityRejected(
  activityId:    string,
  tenantId:      string,
  reviewerNotes: string,
): Promise<void> {
  const activity = await getActivity(activityId, tenantId);
 
  // Mark the current activity as rejected
  await db.update(activities)
    .set({ status: 'rejected', reviewerNotes, rejectedAt: new Date() })
    .where(eq(activities.id, activityId));
 
  // Create a revision activity — same step, same agent, same pipeline position
  const revisionActivity = await db.insert(activities).values({
    tenantId,
    deliverableId:       activity.deliverableId,
    deliverablePeriodId: activity.deliverablePeriodId,
    templateId:          activity.templateId,
    pipelineStep:        activity.pipelineStep,   // same step, not incremented
    name:                `${activity.name} (revision)`,
    assigneeType:        'agent',
    assigneeAgentRole:   activity.assigneeAgentRole,
    isApprovalGate:      false,
    inputRef:            activity.inputRef,        // same input as original
    status:              'created',
  }).returning();
 
  // Enqueue with rejection context — agent will receive both its rejected output
  // and the reviewer's notes in the prompt
  await enqueueActivity(revisionActivity[0], tenantId, activity.assigneeAgentRole, {
    wakeReason:    'rejection',
    rejectionRef:  activity.outputRef,   // MongoDB ObjectId of the rejected output
    reviewerNotes,
  });
}

async function onActivityBlocked(
  activityId:   string,
  tenantId:     string,
  blockerType:  BlockerType,
  blockerMessage: string,
  channelId?:   string,
): Promise<void> {
  // 1. Mark the activity blocked
  await db.update(activities)
    .set({
      status:         'blocked',
      blockerType,
      blockerMessage,
      blockedAt:      new Date(),
    })
    .where(eq(activities.id, activityId));
 
  // 2. Create a human-assigned resolution activity in the DM Portal inbox
  await db.insert(activities).values({
    tenantId,
    name:             `Resolve blocker: ${blockerMessage}`,
    assigneeType:     'human',
    isApprovalGate:   false,
    status:           'created',
    parentActivityId: activityId,   // links back to the blocked activity
    metadata: {
      blockerType,
      blockedActivityId: activityId,
      channelId,
    },
  });
 
  // 3. If channel auth expired, flag the channel record
  if (blockerType === 'channel_auth_expired' && channelId) {
    await db.update(channels)
      .set({ status: 'expired' })
      .where(and(eq(channels.id, channelId), eq(channels.tenantId, tenantId)));
  }
}

async function onBlockerResolved(
  resolutionActivityId: string,
  tenantId:             string,
  resolutionNote:       string,
): Promise<void> {
  const resolution = await getActivity(resolutionActivityId, tenantId);
  const blockedActivityId = resolution.metadata.blockedActivityId;
  const blocked = await getActivity(blockedActivityId, tenantId);
 
  // Mark the blocked activity as ready to resume
  await db.update(activities)
    .set({ status: 'created', blockerType: null, blockerMessage: null })
    .where(eq(activities.id, blockedActivityId));
 
  // Re-enqueue with unblocked wake reason so the agent knows to resume
  await enqueueActivity(blocked, tenantId, blocked.assigneeAgentRole, {
    wakeReason:        'unblocked',
    blockerResolution: resolutionNote,
  });
}

async function onSubtaskCompleted(
  subtaskActivityId: string,
  tenantId:          string,
): Promise<void> {
  const subtask = await getActivity(subtaskActivityId, tenantId);
  if (!subtask.parentActivityId) return;  // not a subtask
 
  const parent = await getActivity(subtask.parentActivityId, tenantId);
  if (parent.status !== 'awaiting_subtask') return;  // parent not waiting
 
  await db.update(activities)
    .set({ status: 'created' })
    .where(eq(activities.id, parent.id));
 
  await enqueueActivity(parent, tenantId, parent.assigneeAgentRole, {
    wakeReason:       'subtask_completed',
    subtaskOutputRef: subtask.outputRef,
    subtaskName:      subtask.name,
  });
}

async function onReviewCompleted(
  reviewActivityId: string,
  tenantId:         string,
  approved:         boolean,
  reviewerFeedback: string,
): Promise<void> {
  const review = await getActivity(reviewActivityId, tenantId);
  const parentId = review.metadata.reviewedActivityId;
  const parent = await getActivity(parentId, tenantId);
 
  await db.update(activities)
    .set({ status: 'created' })
    .where(eq(activities.id, parentId));
 
  await enqueueActivity(parent, tenantId, parent.assigneeAgentRole, {
    wakeReason:       approved ? 'review_approved' : 'review_feedback',
    reviewApproved:   approved,
    reviewerFeedback: reviewerFeedback,
  });
}

// Called from the approvals API route when reviewer submits their decision
// Full implementation in governance-guardrails.md — onApprovalResolved()
// The queue layer's responsibility: enqueue all linked activities correctly
 
async function enqueueApprovalLinkedActivities(
  approvalId:    string,
  tenantId:      string,
  approved:      boolean,
  reviewerNotes: string,
  chosenOption?: string,
): Promise<void> {
  const links = await db.query.approvalLinkedActivities.findMany({
    where: eq(approvalLinkedActivities.approvalId, approvalId),
    with:  { activity: true },
  });
 
  const feedbackText = [reviewerNotes, chosenOption ? `Selected: ${chosenOption}` : null]
    .filter(Boolean).join('\n\n');
 
  // All linked activities re-enqueued in parallel
  await Promise.all(links.map(({ activity }) =>
    enqueueActivity(activity, tenantId, activity.assigneeAgentRole, {
      wakeReason:       approved ? 'review_approved' : 'review_feedback',
      reviewApproved:   approved,
      reviewerFeedback: feedbackText,
    })
  ));
}

async function onActivityReassigned(
  activityId:   string,
  tenantId:     string,
  assigneeType: 'agent' | 'human',
  agentRole:    AgentRole | null,
  humanRole:    string | null,
  reason:       string,
): Promise<void> {
  const activity = await getActivity(activityId, tenantId);
 
  // Mark original as reassigned (terminal — no further processing)
  await db.update(activities)
    .set({ status: 'reassigned', reassignReason: reason })
    .where(eq(activities.id, activityId));
 
  // Create a new activity for the target assignee, same step and input
  const newActivity = await db.insert(activities).values({
    tenantId,
    deliverableId:       activity.deliverableId,
    deliverablePeriodId: activity.deliverablePeriodId,
    templateId:          activity.templateId,
    pipelineStep:        activity.pipelineStep,
    name:                activity.name,
    assigneeType,
    assigneeAgentRole:   agentRole,
    assigneeHumanRole:   humanRole,
    isApprovalGate:      assigneeType === 'human',
    inputRef:            activity.inputRef,
    parentActivityId:    activity.id,       // link back for audit trail
    status:              'created',
    metadata: { reassignedFrom: activityId, reassignReason: reason },
  }).returning();
 
  if (assigneeType === 'agent') {
    await enqueueActivity(newActivity[0], tenantId, agentRole, {
      wakeReason: 'new_task',
    });
  }
  // Human activities surface automatically in DM Portal inbox
}

// Runs at 00:00 on 1st of every month (UTC)
// One job per tenant in the system
await deliverablePeriodQueue.add(
  'start-new-period',
  { tenantId },
  { repeat: { pattern: '0 0 1 * *' } }
);

queue/
├── src/
│   ├── queues.ts          # Queue factory: getQueue(tenantId, agentRole)
│   ├── workers.ts         # Worker factory: createAgentWorker(), processActivity()
│   ├── jobs.ts            # ActivityJobData type definition
│   ├── prompt.ts          # buildActivityPrompt() — threads wake reason + rejection/blocker context
│   ├── pipeline.ts        # onActivityCompleted(), onActivityRejected(), step progression
│   ├── blocker.ts         # onActivityBlocked(), onBlockerResolved()
│   ├── subtask.ts         # onSubtaskCompleted()
│   ├── review.ts          # onReviewCompleted()
│   ├── approval.ts        # enqueueApprovalLinkedActivities() — fan-out on approval resolve
│   ├── reassign.ts        # onActivityReassigned()
│   ├── scheduler.ts       # Monthly period cron registration
│   └── deadletter.ts      # Dead letter handler
└── package.json