task-context.ts

/**
 * Task Context — Access execution context from within a worker
 *
 * Demonstrates TaskContext for logging, callbacks, and introspection:
 *   - getTaskContext() to access the current task's context
 *   - addLog() to append execution logs visible in Conductor UI
 *   - setCallbackAfter() for long-running tasks (IN_PROGRESS pattern)
 *   - getInput(), setOutput() for data access
 *   - getRetryCount(), getPollCount() for introspection
 *
 * Run:
 *   CONDUCTOR_SERVER_URL=http://localhost:8080 npx ts-node examples/task-context.ts
 */
import {
  OrkesClients,
  ConductorWorkflow,
  TaskHandler,
  worker,
  simpleTask,
  getTaskContext,
} from "../src/sdk";
import type { Task } from "../src/open-api";

// ── Worker with context usage ───────────────────────────────────────
const _loggingWorker = worker({ taskDefName: "ctx_logging_worker", registerTaskDef: true })(
  async (task: Task) => {
    const ctx = getTaskContext();

    // Log execution metadata
    ctx?.addLog(`Task started — taskId=${ctx.getTaskId()}`);
    ctx?.addLog(`Workflow: ${ctx.getWorkflowInstanceId()}`);
    ctx?.addLog(`Retry count: ${ctx.getRetryCount()}`);
    ctx?.addLog(`Poll count: ${ctx.getPollCount()}`);
    ctx?.addLog(`Task def: ${ctx.getTaskDefName()}`);

    // Access input through context
    const input = ctx?.getInput() ?? task.inputData ?? {};
    ctx?.addLog(`Input keys: ${Object.keys(input).join(", ")}`);

    // Set output through context
    ctx?.setOutput({
      processed: true,
      taskId: ctx.getTaskId(),
      message: input.message ?? "no message",
    });

    ctx?.addLog("Task completed successfully");

    return {
      status: "COMPLETED",
      outputData: ctx?.getOutput() ?? { processed: true },
    };
  }
);

// ── Worker demonstrating IN_PROGRESS callback pattern ───────────────
const _longRunningWorker = worker({ taskDefName: "ctx_long_running", registerTaskDef: true })(
  async (task: Task) => {
    const ctx = getTaskContext();
    const attempt = (task.inputData?.attempt as number) ?? 0;

    ctx?.addLog(`Long-running task attempt ${attempt}`);

    // Simulate a task that needs multiple callbacks
    if (attempt < 2) {
      ctx?.addLog("Not ready yet, requesting callback in 5 seconds");
      ctx?.setCallbackAfter(5);
      return {
        status: "IN_PROGRESS" as const,
        callbackAfterSeconds: 5,
        outputData: { attempt: attempt + 1, progress: (attempt + 1) * 33 },
      };
    }

    ctx?.addLog("Processing complete!");
    return {
      status: "COMPLETED" as const,
      outputData: { result: "done", totalAttempts: attempt + 1 },
    };
  }
);

// ── Main ────────────────────────────────────────────────────────────
async function main() {
  const clients = await OrkesClients.from();
  const workflowClient = clients.getWorkflowClient();
  const client = clients.getClient();

  const wf = new ConductorWorkflow(workflowClient, "task_context_example")
    .description("Demonstrates TaskContext features")
    .add(
      simpleTask("log_ref", "ctx_logging_worker", {
        message: "${workflow.input.message}",
      })
    )
    .outputParameters({
      taskId: "${log_ref.output.taskId}",
      message: "${log_ref.output.message}",
    });

  await wf.register(true);
  console.log("Registered workflow:", wf.getName());

  const handler = new TaskHandler({ client, scanForDecorated: true });
  await handler.startWorkers();

  const run = await wf.execute({ message: "Hello from context example" });
  console.log("Status:", run.status);
  console.log("Output:", JSON.stringify(run.output, null, 2));

  // Fetch task execution logs
  const taskClient = clients.getTaskClient();
  const tasks = run.tasks ?? [];
  for (const t of tasks) {
    if (t.taskId) {
      const logs = await taskClient.getTaskLogs(t.taskId);
      if (logs.length > 0) {
        console.log(`\nLogs for task ${t.taskDefName}:`);
        for (const log of logs) {
          console.log(`  ${log.log}`);
        }
      }
    }
  }

  await handler.stopWorkers();
  process.exit(0);
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});