mirror of
https://github.com/hanzoai/base.git
synced 2026-10-21 03:41:50 +00:00
zeekay
bb905e0226
Adds native Temporal SDK integration to the tasks plugin: - durable.go: DurableStore client (submit/cancel/signal/status workflows) - workflows.go: AgentTaskWorkflow, PipelineWorkflow, FanOutWorkflow - activities.go: ExecuteTaskActivity with heartbeat + pluggable executor - worker.go: Embedded Temporal worker (polls task queue in-process) SQLite remains authoritative state. Temporal provides crash-safe execution that survives restarts. When durable is enabled: - Task create → SQLite + Temporal workflow submission - Task cancel → SQLite + Temporal workflow cancellation - Task signal → forwarded to Temporal workflow Multi-tenant via namespace-per-org (TASKS_NAMESPACE=org-acme). Env config: TASKS_ENABLED, TASKS_ADDRESS, TASKS_NAMESPACE, TASKS_QUEUE. Default cloud: tasks.hanzo.ai:7233. Falls back to SQLite-only if unreachable.
48 lines
1.3 KiB
Go
48 lines
1.3 KiB
Go
package tasks
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
|
|
"go.temporal.io/sdk/activity"
|
|
)
|
|
|
|
// ExecuteTaskActivity is the Temporal activity that executes a task.
|
|
// Delegates to the Config.OnExecute callback registered during plugin setup.
|
|
//
|
|
// If no executor is configured, the task completes with an acknowledgment message.
|
|
var taskExecutor func(ctx context.Context, task *Task) (*Task, error)
|
|
|
|
// SetTaskExecutor registers the function called by Temporal activities.
|
|
// This is typically set during plugin registration via Config.OnDurableExecute.
|
|
func SetTaskExecutor(fn func(ctx context.Context, task *Task) (*Task, error)) {
|
|
taskExecutor = fn
|
|
}
|
|
|
|
// ExecuteTaskActivity is the activity implementation called by Temporal workflows.
|
|
func ExecuteTaskActivity(ctx context.Context, task *Task) (*Task, error) {
|
|
logger := activity.GetLogger(ctx)
|
|
logger.Info("Executing task",
|
|
"task_id", task.ID,
|
|
"space_id", task.SpaceID,
|
|
"title", task.Title,
|
|
"assigned_to", task.AssignedTo,
|
|
)
|
|
|
|
if taskExecutor == nil {
|
|
task.State = TaskCompleted
|
|
task.Output = map[string]any{"message": "No executor registered. Task acknowledged."}
|
|
return task, nil
|
|
}
|
|
|
|
// Heartbeat so Temporal knows we're alive.
|
|
activity.RecordHeartbeat(ctx, "executing")
|
|
|
|
result, err := taskExecutor(ctx, task)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("task execution failed: %w", err)
|
|
}
|
|
|
|
return result, nil
|
|
}
|