// Example: External service integrating with Poimen RoutingWorkflow // Shows how to submit a task and wait for completion package main import ( "context" "encoding/json" "fmt" "log" "time" "go.temporal.io/sdk/client" ) // ---- Types (mirror internal/routing/types.go) ---- type WorkflowSpec struct { Name string `json:"name"` Input map[string]interface{} `json:"input,omitempty"` States []State `json:"states"` } type State struct { Name string `json:"name"` Type string `json:"type"` // Task, Pass, Fail Resource string `json:"resource,omitempty"` Parameters map[string]interface{} `json:"parameters,omitempty"` Timeout string `json:"timeout,omitempty"` Next string `json:"next,omitempty"` End bool `json:"end,omitempty"` } type RoutingWorkflowInput struct { Spec *WorkflowSpec `json:"spec"` } type RoutingWorkflowOutput struct { Status string `json:"status"` StepResults map[string]map[string]interface{} `json:"stepResults"` Error string `json:"error,omitempty"` } // ---- Example Service ---- type TaskService struct { temporalClient client.Client taskQueue string } func NewTaskService(temporalHost, namespace, taskQueue string) (*TaskService, error) { c, err := client.Dial(client.Options{ HostPort: temporalHost, Namespace: namespace, }) if err != nil { return nil, fmt.Errorf("failed to connect to Temporal: %w", err) } return &TaskService{ temporalClient: c, taskQueue: taskQueue, }, nil } func (s *TaskService) Close() { s.temporalClient.Close() } // SubmitAndWait submits a workflow spec and waits for completion func (s *TaskService) SubmitAndWait(ctx context.Context, spec *WorkflowSpec, timeout time.Duration) (*RoutingWorkflowOutput, error) { workflowID := fmt.Sprintf("%s-%d", spec.Name, time.Now().UnixNano()) // Start workflow run, err := s.temporalClient.ExecuteWorkflow(ctx, client.StartWorkflowOptions{ ID: workflowID, TaskQueue: s.taskQueue, }, "RoutingWorkflow", RoutingWorkflowInput{Spec: spec}) if err != nil { return nil, fmt.Errorf("failed to start workflow: %w", err) } log.Printf("Workflow started: ID=%s, RunID=%s", run.GetID(), run.GetRunID()) // Wait for completion with timeout waitCtx, cancel := context.WithTimeout(ctx, timeout) defer cancel() var result RoutingWorkflowOutput if err := run.Get(waitCtx, &result); err != nil { return nil, fmt.Errorf("workflow failed: %w", err) } return &result, nil } // SubmitAsync submits workflow and returns immediately (fire-and-forget) func (s *TaskService) SubmitAsync(ctx context.Context, spec *WorkflowSpec) (workflowID string, runID string, err error) { workflowID = fmt.Sprintf("%s-%d", spec.Name, time.Now().UnixNano()) run, err := s.temporalClient.ExecuteWorkflow(ctx, client.StartWorkflowOptions{ ID: workflowID, TaskQueue: s.taskQueue, }, "RoutingWorkflow", RoutingWorkflowInput{Spec: spec}) if err != nil { return "", "", fmt.Errorf("failed to start workflow: %w", err) } return run.GetID(), run.GetRunID(), nil } // WaitForCompletion waits for an existing workflow to complete func (s *TaskService) WaitForCompletion(ctx context.Context, workflowID string, timeout time.Duration) (*RoutingWorkflowOutput, error) { run := s.temporalClient.GetWorkflow(ctx, workflowID, "") waitCtx, cancel := context.WithTimeout(ctx, timeout) defer cancel() var result RoutingWorkflowOutput if err := run.Get(waitCtx, &result); err != nil { return nil, fmt.Errorf("workflow failed: %w", err) } return &result, nil } // GetStatus gets current workflow status without waiting func (s *TaskService) GetStatus(ctx context.Context, workflowID string) (string, error) { desc, err := s.temporalClient.DescribeWorkflowExecution(ctx, workflowID, "") if err != nil { return "", err } return desc.WorkflowExecutionInfo.Status.String(), nil } // ---- Example Usage ---- func main() { // Connect to Temporal svc, err := NewTaskService( "temporal-frontend.temporal:7233", "poimen-harness", "poimen-taskqueue", ) if err != nil { log.Fatal(err) } defer svc.Close() ctx := context.Background() // Example 1: Implement task T0.3 with code analysis fmt.Println("=== Example 1: Submit and Wait ===") spec := &WorkflowSpec{ Name: "implement-T0.3", Input: map[string]interface{}{ "taskId": "T0.3", "description": "Implement git worktree management", "repo": "https://github.com/rockliang/poimen", }, States: []State{ { Name: "Clone", Type: "Task", Resource: "CloneRepoActivity", Parameters: map[string]interface{}{ "repo": "${workflow.input.repo}", "branch": "main", }, Timeout: "5m", Next: "Analyze", }, { Name: "Analyze", Type: "Task", Resource: "AnalyzeCodeActivity", Parameters: map[string]interface{}{ "path": "${Clone.output.path}", "depth": 3, }, Timeout: "10m", Next: "Report", }, { Name: "Report", Type: "Task", Resource: "GenerateReportActivity", Parameters: map[string]interface{}{ "analysisResult": "${Analyze.output}", "format": "markdown", }, Timeout: "2m", End: true, }, }, } result, err := svc.SubmitAndWait(ctx, spec, 30*time.Minute) if err != nil { log.Printf("Error: %v", err) } else { fmt.Printf("Status: %s\n", result.Status) for step, output := range result.StepResults { fmt.Printf(" %s: %v\n", step, output) } } // Example 2: Fire and forget, then poll fmt.Println("\n=== Example 2: Async Submit + Poll ===") workflowID, runID, err := svc.SubmitAsync(ctx, spec) if err != nil { log.Fatal(err) } fmt.Printf("Submitted: workflowID=%s, runID=%s\n", workflowID, runID) // Poll status for i := 0; i < 5; i++ { status, _ := svc.GetStatus(ctx, workflowID) fmt.Printf(" Poll %d: status=%s\n", i+1, status) if status == "WORKFLOW_EXECUTION_STATUS_COMPLETED" { break } time.Sleep(5 * time.Second) } // Get final result result, err = svc.WaitForCompletion(ctx, workflowID, 30*time.Minute) if err != nil { log.Printf("Error: %v", err) } else { resultJSON, _ := json.MarshalIndent(result, "", " ") fmt.Printf("Final result:\n%s\n", resultJSON) } }