Orchestration concepts

Task orchestration is how a Teamwork Graph connector keeps data fresh without an external scheduler. The platform invokes a function in your connector on a fixed cadence per connection, and your code can fan that single tick out into smaller per-item invocations that the platform delivers asynchronously. This guide covers the concepts you need before you wire orchestration into a connector: the IDs the platform passes you, how task types work, how to make handlers idempotent, how to opt into long-running execution, and the lifecycle rules that govern when the platform stops invoking your code.

For a step-by-step walkthrough of wiring orchestration into a Forge connector (with a Google Drive worked example), see Build an orchestrated connector. For the SDK methods that drive each step, see Task operations.

How the methods fit together

A typical connector uses the four task operations in this order:

1. From onConnectionChange, call scheduleOrUpdateTask once per root task you want on the connection. The same call is safe on both CREATED and UPDATED events because re-calling with the same taskId either creates the schedule or updates it in place.
2. The platform then invokes your taskRunner on the configured cadence. Each invocation receives a payload with connectionId, scanId, taskId, taskExecutionId, and your connectorFormFields.
3. Inside taskRunner, do the actual work (call setObjects, setUsers, etc.). For larger work, call scheduleChildTask to fan out into smaller per-item invocations that the platform delivers back to the same taskRunner.
4. Before returning from taskRunner, call updateTaskStatus with 'success' or 'failure' (plus a failureReason when failing). The platform uses this to surface task health in the connector UI and decide whether to retry.

For a complete worked example, see the Google Drive connector example app.

The three IDs: taskId, scanId, taskExecutionId

Every TaskRunnerPayload carries three identifiers that look similar but mean different things.

A root tick that fans out to two children produces a tree like:

1
2
scanId = S1

  taskId = R    (root)
   |
   |- taskId = C1   (child)
   |- taskId = C2   (child)

updateTaskStatus always identifies the call by (scanId, taskExecutionId, taskId). Pass them through unchanged from the current TaskRunnerPayload.

taskId must be a UUID

scheduleOrUpdateTask and scheduleChildTask both require task.taskId to be a valid UUID string. Free-form strings such as "my-app:gdrive:user-sync" will be rejected client-side with task.taskId must be a valid UUID format.

The recommended pattern depends on whether the task is a root or a child:

• Root tasks: mint a random UUID once with uuid() and persist the mapping (for example under a key like rootTaskId:<connectionId>:<taskType>) so re-running the registration flow reuses the same taskId. This makes scheduleOrUpdateTask an update of the existing schedule rather than a competing schedule.
• Child tasks: derive the UUID deterministically from a stable per-child key using uuidv5, for example uuidv5("<scanId>:<parentTaskId>:<itemId>", APP_NAMESPACE). The platform schedules a fresh task execution for every scheduleChildTask call regardless of the taskId you pass; deterministic IDs are how your runner recognizes a redelivery. When the parent is retried and re-derives the same child taskId, every redelivered child invocation looks up the same TaskInfo row in KVS and the completed: true short-circuit (see At-least-once delivery and idempotency) absorbs the duplicate. With random UUIDs, each duplicated child has a fresh taskId and a fresh TaskInfo, so your runner cannot tell duplicates from real work and runs the full pipeline every time.

Task types

taskType is a closed enum exported as types.FORGE_TASK_TYPES. Allowed values match the cross product of {user, group, entity, relationship} x {full, incremental}:

1
2
import { types } from '@forge/teamwork-graph';

types.FORGE_TASK_TYPES.USER_INGESTION_FULL;             // 'user_ingestion_full'
types.FORGE_TASK_TYPES.USER_INGESTION_INCREMENTAL;      // 'user_ingestion_incremental'
types.FORGE_TASK_TYPES.GROUP_INGESTION_FULL;            // 'group_ingestion_full'
types.FORGE_TASK_TYPES.GROUP_INGESTION_INCREMENTAL;     // 'group_ingestion_incremental'
types.FORGE_TASK_TYPES.ENTITY_INGESTION_FULL;           // 'entity_ingestion_full'
types.FORGE_TASK_TYPES.ENTITY_INGESTION_INCREMENTAL;    // 'entity_ingestion_incremental'
types.FORGE_TASK_TYPES.RELATIONSHIP_INGESTION_FULL;     // 'relationship_ingestion_full'
types.FORGE_TASK_TYPES.RELATIONSHIP_INGESTION_INCREMENTAL; // 'relationship_ingestion_incremental'

Submitting any other string is rejected client-side by the SDK with task.taskType must be one of: user_ingestion_full, user_ingestion_incremental, ....

Use the same FORGE_TASK_TYPES value as your dispatch key inside taskRunner and persist it on the TaskInfo you write to KVS. There is no need for a parallel internal enum:

1
2
import { graph, types } from '@forge/teamwork-graph';

switch (taskInfo.taskType) {
  case types.FORGE_TASK_TYPES.ENTITY_INGESTION_INCREMENTAL:
    return runEntityIngestion(request, taskInfo);
  case types.FORGE_TASK_TYPES.USER_INGESTION_INCREMENTAL:
    return runUserSync(request);
  default:
    throw new Error(`Unsupported task type: ${taskInfo.taskType}`);
}

Child tasks typically share their parent's taskType (the same taskRunner routine handles the root tick and the per-item children). Use TaskInfo.metadata to carry the per-invocation context that distinguishes them, for example { folderId: 'abc-123' } on a folder-walk child versus the root folder on the parent.

TaskInfo persistence

The TaskRunnerPayload your handler receives carries identifiers (taskId, scanId, taskExecutionId, connectionId) and connectorFormFields, but it does not reliably carry the work-specific context you need to dispatch correctly.

The SDK's ChildTask type accepts two free-form metadata fields (temporalTaskMetadata and persistentTaskMetadata), but propagation of these payloads back to your taskRunner as TaskRunnerPayload.taskMetadata is not a stable contract today. Treat them as platform-only inputs.

The reliable channel is to persist a small TaskInfo record in KVS keyed by taskId before you schedule, and read it back inside taskRunner:

1
2
import { kvs } from '@forge/kvs';
import { types } from '@forge/teamwork-graph';

interface TaskInfo {
  taskType: types.ForgeTaskType;
  connectionId: string;
  createdAt: string;
  metadata?: Record<string, any>;
  completed?: boolean;
}

async function setTaskInfo(taskId: string, info: TaskInfo): Promise<void> {
  await kvs.set(`taskInfo:${taskId}`, info);
}

async function getTaskInfo(taskId: string): Promise<TaskInfo | undefined> {
  return (await kvs.get(`taskInfo:${taskId}`)) as TaskInfo | undefined;
}

Write TaskInfo before calling scheduleChildTask, read it as the first step of taskRunner, and use its taskType to dispatch.

At-least-once delivery and idempotency

The platform delivers each task to your taskRunner at least once and does not currently deduplicate child schedules by taskId server-side. Your handler must be idempotent: a second delivery of an already-completed task should not double the work or report a spurious failure. Pair the runner-side completed: true check below with deterministic child taskIds (see taskId must be a UUID) so duplicate deliveries land on the same KVS row this check reads.

The standard pattern is:

1. Read TaskInfo keyed by taskId at the top of taskRunner.
2. If TaskInfo.completed is true, call updateTaskStatus({ status: 'success' }) and return.
3. Otherwise, run the work, then write { ...taskInfo, completed: true } back to KVS and call updateTaskStatus({ status: 'success' }).

1
2
export async function taskRunner(request: TaskRunnerPayload): Promise<TaskRunnerResult> {
  const taskInfo = await getTaskInfo(request.taskId);
  if (!taskInfo) {
    await graph.updateTaskStatus({
      scanId: request.scanId,
      taskExecutionId: request.taskExecutionId,
      connectionId: request.connectionId,
      status: 'failure',
      failureReason: 'ENTITY_NOT_FOUND',
      task: { taskId: request.taskId },
    });
    return { success: false, message: 'Task not found' };
  }

  if (taskInfo.completed) {
    await graph.updateTaskStatus({
      scanId: request.scanId,
      taskExecutionId: request.taskExecutionId,
      connectionId: request.connectionId,
      status: 'success',
      task: { taskId: request.taskId },
    });
    return { success: true, message: 'Already completed (duplicate delivery)' };
  }

  // ...run the work, mark completed, then call updateTaskStatus({ status: 'success' })...
}

A common anti-pattern is deleting TaskInfo from KVS as soon as the work succeeds. The duplicate invocation then reads undefined, reports ENTITY_NOT_FOUND, and overwrites the earlier success. Keep the record around with a completed: true marker; clean it up later in a maintenance task or in onConnectionChange on DELETED.

For child tasks, pair this with deterministic UUIDs (see taskId must be a UUID) so every redelivered child resolves to the same TaskInfo row in KVS and the completed: true check above can absorb it. The platform itself does not deduplicate child schedules by taskId today; this runner-side check is where deduplication happens.

Long-running tasks (up to 15 minutes)

The default per-invocation execution budget for a taskRunner is in the tens of seconds. To opt into a 15-minute budget, set timeoutSeconds: 900 on the function entry that backs your orchestration.taskRunner:

1
2
modules:
  graph:connector:
    - key: my-connector
      orchestration:
        taskRunner:
          function: taskRunnerFn
  function:
    - key: taskRunnerFn
      handler: index.taskRunner
      timeoutSeconds: 900

timeoutSeconds is only valid on a function that is referenced by a consumer module. Setting it on an unreferenced function fails manifest validation with Timeout is not applicable for function X: not referenced by a consumer module.

Even with the extended budget, prefer fanning out via scheduleChildTask over running a single long handler. Smaller children survive transient failures better, retry independently, and let the platform parallelize work across invocations.

Lifecycle: when the platform stops invoking your taskRunner

The platform stops invoking your taskRunner for a given connection or task in any of these situations:

• The connection is deleted. Your onConnectionChange handler should also remove any KVS state keyed by connectionId (root taskId mappings, TaskInfo rows for in-flight tasks).
• The app is uninstalled.
• The manifest no longer wires orchestration.taskRunner and you redeploy.
• A task is marked failed with a failureReason. The platform decides retry behavior from the reason; the connector reports the reason via updateTaskStatus and lets the platform handle escalation. See updateTaskStatus.

Active schedules are not removed on redeploy on their own. As long as manifest.yml keeps wiring orchestration.taskRunner and the connection stays in place, the platform keeps invoking the runner on the existing cadence.

scheduleOrUpdateTask update semantics

Calling scheduleOrUpdateTask with the same taskId updates the existing schedule, including the cadence. Changing scheduleInterval.value or scheduleInterval.timeUnit takes effect from the next tick. The first invocation after a fresh scheduleOrUpdateTask typically fires within a few seconds, then the configured interval governs subsequent invocations. See scheduleOrUpdateTask.

Common gotchas

• scheduleOrUpdateTask returns { status: 'ACCEPTED' } and my code reports failure. Branch on response.error, not on response.success. The success path of orchestration responses does not set a boolean success field.
• task.taskType must be one of: ... from the SDK. taskType is a closed enum. Use types.FORGE_TASK_TYPES.* constants from the SDK; arbitrary strings are rejected client-side.
• task.taskId must be a valid UUID format. Use uuid() for root tasks and uuidv5(...) for children. Persist root mappings to KVS so re-runs reuse the same id.
• scheduleInterval.value must be between 1 and 60. The cadence value is bounded; the unit controls whether it counts minutes, hours, or days.
• Failed to schedule child tasks: Not Found - Scan not found. The scanId you passed does not match an active scan on the connection. This usually means the call is happening outside of an active taskRunner invocation, or against a stale scanId.
• My handler reports ENTITY_NOT_FOUND after a successful run. You are deleting TaskInfo too eagerly. Switch to a completed: true flag and short-circuit duplicate invocations to a no-op success.
• taskMetadata is empty in my taskRunner. Whatever you set on scheduleChildTask's temporalTaskMetadata or persistentTaskMetadata is not a stable channel back to the runner. Stash everything your handler needs in KVS via TaskInfo.metadata.